1 /*

     2  * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.jfr.consumer;

    27 

    28 import java.io.IOException;

    29 import java.nio.file.Path;

    30 import java.security.AccessControlContext;

    31 import java.security.AccessController;

    32 import java.time.Duration;

    33 import java.time.Instant;

    34 import java.util.Objects;

    35 import java.util.function.Consumer;

    36 

    37 import jdk.jfr.internal.SecuritySupport;

    38 import jdk.jfr.internal.Utils;

    39 import jdk.jfr.internal.consumer.EventDirectoryStream;

    40 import jdk.jfr.internal.consumer.EventFileStream;

    41 import jdk.jfr.internal.consumer.FileAccess;

    42 

    43 /**

    44  * Represents a stream of events.

    45  * <p>

    46  * A stream is a sequence of events and the way to interact with a stream is to

    47  * register actions. The {@code EventStream} interface is not to be implemented

    48  * and future versions of the JDK may prevent this completely.

    49  * <p>

    50  * To receive a notification when an event arrives, register an action using the

    51  * {@link #onEvent(Consumer)} method. To filter the stream for an event with a

    52  * specific name, use {@link #onEvent(String, Consumer)} method.

    53  * <p>

    54  * By default, the same {@code RecordedEvent} object can be used to

    55  * represent two or more distinct events. That object can be delivered

    56  * multiple times to the same action as well as to other actions. To use an

    57  * event object after the action is completed, the

    58  * {@link #setReuse(boolean)} method should be set to {@code false} so a

    59  * new object is allocated for each event.

    60  * <p>

    61  * Events are delivered in batches. To receive a notification when a batch is

    62  * complete, register an action using the {@link #onFlush(Runnable)} method.

    63  * This is an opportunity to aggregate or push data to external systems while

    64  * the Java Virtual Machine (JVM) is preparing the next batch.

    65  * <p>

    66  * Events within a batch are sorted chronologically by their end time.

    67  * Well-ordering of events is only maintained for events available to the JVM at

    68  * the point of flush, i.e. for the set of events delivered as a unit in a

    69  * single batch. Events delivered in a batch could therefore be out-of-order

    70  * compared to events delivered in a previous batch, but never out-of-order with

    71  * events within the same batch. If ordering is not a concern, sorting can be

    72  * disabled using the {@link #setOrdered(boolean)} method.

    73  * <p>

    74  * To dispatch events to registered actions, the stream must be started. To

    75  * start processing in the current thread, invoke the {@link #start()} method.

    76  * To process actions asynchronously in a separate thread, invoke the

    77  * {@link #startAsync()} method. To await completion of the stream, use the

    78  * awaitTermination {@link #awaitTermination()} or the

    79  * {@link #awaitTermination(Duration)} method.

    80  * <p>

    81  * When a stream ends it is automatically closed. To manually stop processing of

    82  * events, close the stream by invoking the {@link #close()} method. A stream

    83  * can also be automatically closed in exceptional circumstances, for example if

    84  * the JVM that is being monitored exits. To receive a notification in any of

    85  * these occasions, use the {@link #onClose(Runnable)} method to register an

    86  * action.

    87  * <p>

    88  * If an unexpected exception occurs in an action, it is possible to catch the

    89  * exception in an error handler. An error handler can be registered using the

    90  * {@link #onError(Runnable)} method. If no error handler is registered, the

    91  * default behavior is to print the exception and its backtrace to the standard

    92  * error stream.

    93  * <p>

    94  * The following example shows how an {@code EventStream} can be used to listen

    95  * to events on a JVM running Flight Recorder

    96  *

    97  * <pre>

    98  * <code>

    99  * try (var es = EventStream.openRepository()) {

   100  *   es.onEvent("jdk.CPULoad", event -> {

   101  *     System.out.println("CPU Load " + event.getEndTime());

   102  *     System.out.println(" Machine total: " + 100 * event.getFloat("machineTotal") + "%");

   103  *     System.out.println(" JVM User: " + 100 * event.getFloat("jvmUser") + "%");

   104  *     System.out.println(" JVM System: " + 100 * event.getFloat("jvmSystem") + "%");

   105  *     System.out.println();

   106  *   });

   107  *   es.onEvent("jdk.GarbageCollection", event -> {

   108  *     System.out.println("Garbage collection: " + event.getLong("gcId"));

   109  *     System.out.println(" Cause: " + event.getString("cause"));

   110  *     System.out.println(" Total pause: " + event.getDuration("sumOfPauses"));

   111  *     System.out.println(" Longest pause: " + event.getDuration("longestPause"));

   112  *     System.out.println();

   113  *   });

   114  *   es.start();

   115  * }

   116  * </code>

   117  * </pre>

   118  * <p>

   119  * To start recording together with the stream, see {@link RecordingStream}.

   120  *

   121  * @since 14

   122  */

   123 public interface EventStream extends AutoCloseable {

   124     /**

   125      * Creates a stream from the repository of the current Java Virtual Machine

   126      * (JVM).

   127      * <p>

   128      * By default, the stream starts with the next event flushed by Flight

   129      * Recorder.

   130      *

   131      * @return an event stream, not {@code null}

   132      *

   133      * @throws IOException if a stream can't be opened, or an I/O error occurs

   134      *         when trying to access the repository

   135      *

   136      * @throws SecurityException if a security manager exists and the caller

   137      *         does not have

   138      *         {@code FlightRecorderPermission("accessFlightRecorder")}

   139      */

   140     public static EventStream openRepository() throws IOException {

   141         Utils.checkAccessFlightRecorder();

   142         return new EventDirectoryStream(AccessController.getContext(), null, SecuritySupport.PRIVILIGED, false);

   143     }

   144 

   145     /**

   146      * Creates an event stream from a disk repository.

   147      * <p>

   148      * By default, the stream starts with the next event flushed by Flight

   149      * Recorder.

   150      *

   151      * @param directory location of the disk repository, not {@code null}

   152      *

   153      * @return an event stream, not {@code null}

   154      *

   155      * @throws IOException if a stream can't be opened, or an I/O error occurs

   156      *         when trying to access the repository

   157      *

   158      * @throws SecurityException if a security manager exists and its

   159      *         {@code checkRead} method denies read access to the directory, or

   160      *         files in the directory.

   161      */

   162     public static EventStream openRepository(Path directory) throws IOException {

   163         Objects.nonNull(directory);

   164         AccessControlContext acc = AccessController.getContext();

   165         return new EventDirectoryStream(acc, directory, FileAccess.UNPRIVILIGED, false);

   166     }

   167 

   168     /**

   169      * Creates an event stream from a file.

   170      * <p>

   171      * By default, the stream starts with the first event in the file.

   172      *

   173      * @param file location of the file, not {@code null}

   174      *

   175      * @return an event stream, not {@code null}

   176      *

   177      * @throws IOException if the file can't be opened, or an I/O error occurs

   178      *         during reading

   179      *

   180      * @throws SecurityException if a security manager exists and its

   181      *         {@code checkRead} method denies read access to the file

   182      */

   183     static EventStream openFile(Path file) throws IOException {

   184         return new EventFileStream(AccessController.getContext(), file);

   185     }

   186 

   187     /**

   188      * Registers an action to perform on all events in the stream.

   189      *

   190      * @param action an action to perform on each {@code RecordedEvent}, not

   191      *        {@code null}

   192      */

   193     void onEvent(Consumer<RecordedEvent> action);

   194 

   195     /**

   196      * Registers an action to perform on all events matching a name.

   197      *

   198      * @param eventName the name of the event, not {@code null}

   199      *

   200      * @param action an action to perform on each {@code RecordedEvent} matching

   201      *        the event name, not {@code null}

   202      */

   203     void onEvent(String eventName, Consumer<RecordedEvent> action);

   204 

   205     /**

   206      * Registers an action to perform after the stream has been flushed.

   207      *

   208      * @param action an action to perform after the stream has been

   209      *        flushed, not {@code null}

   210      */

   211     void onFlush(Runnable action);

   212 

   213     /**

   214      * Registers an action to perform if an exception occurs.

   215      * <p>

   216      * if an action is not registered, an exception stack trace is printed to

   217      * standard error.

   218      * <p>

   219      * Registering an action overrides the default behavior. If multiple actions

   220      * have been registered, they are performed in the order of registration.

   221      * <p>

   222      * If this method itself throws an exception, resulting behavior is

   223      * undefined.

   224      *

   225      * @param action an action to perform if an exception occurs, not

   226      *        {@code null}

   227      */

   228     void onError(Consumer<Throwable> action);

   229 

   230     /**

   231      * Registers an action to perform when the stream is closed.

   232      * <p>

   233      * If the stream is already closed, the action will be performed immediately

   234      * in the current thread.

   235      *

   236      * @param action an action to perform after the stream is closed, not

   237      *        {@code null}

   238      * @see #close()

   239      */

   240     void onClose(Runnable action);

   241 

   242     /**

   243      * Releases all resources associated with this stream.

   244      * <p>

   245      * Closing a previously closed stream has no effect.

   246      */

   247     void close();

   248 

   249     /**

   250      * Unregisters an action.

   251      * <p>

   252      * If the action has been registered multiple times, all instances are

   253      * unregistered.

   254      *

   255      * @param action the action to unregister, not {@code null}

   256      *

   257      * @return {@code true} if the action was unregistered, {@code false}

   258      *         otherwise

   259      *

   260      * @see #onEvent(Consumer)

   261      * @see #onEvent(String, Consumer)

   262      * @see #onFlush(Runnable)

   263      * @see #onClose(Runnable)

   264      * @see #onError(Consumer)

   265      */

   266     boolean remove(Object action);

   267 

   268     /**

   269      * Specifies that the event object in an {@link #onEvent(Consumer)} action

   270      * can be reused.

   271      * <p>

   272      * If reuse is set to {@code true), an action should not keep a reference

   273      * to the event object after the action has completed.

   274      *

   275      * @param reuse {@code true} if an event object can be reused, {@code false}

   276      * otherwise

   277      */

   278     void setReuse(boolean reuse);

   279 

   280     /**

   281      * Specifies that events arrives in chronological order, sorted by the time

   282      * they were committed to the stream.

   283      *

   284      * @param ordered if event objects arrive in chronological order to

   285      *        {@code #onEvent(Consumer)}

   286      */

   287     void setOrdered(boolean ordered);

   288 

   289     /**

   290      * Specifies the start time of the stream.

   291      * <p>

   292      * The start time must be set before starting the stream

   293      *

   294      * @param startTime the start time, not {@code null}

   295      *

   296      * @throws IllegalStateException if the stream is already started

   297      *

   298      * @see #start()

   299      * @see #startAsync()

   300      */

   301     void setStartTime(Instant startTime);

   302 

   303     /**

   304      * Specifies the end time of the stream.

   305      * <p>

   306      * The end time must be set before starting the stream.

   307      * <p>

   308      * At end time, the stream is closed.

   309      *

   310      * @param endTime the end time, not {@code null}

   311      *

   312      * @throws IllegalStateException if the stream is already started

   313      *

   314      * @see #start()

   315      * @see #startAsync()

   316      */

   317     void setEndTime(Instant endTime);

   318 

   319     /**

   320      * Start processing of actions.

   321      * <p>

   322      * Actions are performed in the current thread.

   323      *

   324      * @throws IllegalStateException if the stream is already started or closed

   325      */

   326     void start();

   327 

   328     /**

   329      * Start asynchronous processing of actions.

   330      * <p>

   331      * Actions are performed in a single separate thread.

   332      *

   333      * @throws IllegalStateException if the stream is already started or closed

   334      */

   335     void startAsync();

   336 

   337     /**

   338      * Blocks until all actions are completed, or the stream is closed, or the

   339      * timeout occurs, or the current thread is interrupted, whichever happens

   340      * first.

   341      *

   342      * @param timeout the maximum time to wait, not {@code null}

   343      *

   344      * @throws IllegalArgumentException if timeout is negative

   345      * @throws InterruptedException if interrupted while waiting

   346      *

   347      * @see #start()

   348      * @see #startAsync()

   349      * @see Thread#interrupt()

   350      */

   351     void awaitTermination(Duration timeout) throws InterruptedException;

   352 

   353     /**

   354      * Blocks until all actions are completed, or the stream is closed, or the

   355      * current thread is interrupted, whichever happens first.

   356      *

   357      * @throws InterruptedException if interrupted while waiting

   358      *

   359      * @see #start()

   360      * @see #startAsync()

   361      * @see Thread#interrupt()

   362      */

   363     void awaitTermination() throws InterruptedException;

   364 }