InterceptorFactory.java

package com.jinspired.jxinsight.probes.neo4j.gtrace;

import org.jinspired.probes.Probes;
import org.jinspired.probes.interceptor.ProbesInterceptor;
import org.jinspired.probes.interceptor.ProbesInterceptorFactory;
import org.neo4j.graphdb.*;
import org.neo4j.graphdb.factory.GraphDatabaseFactory;
import org.neo4j.graphdb.schema.Schema;

import java.util.HashMap;

import static org.jinspired.probes.Probes.name;
import static org.neo4j.graphdb.DynamicLabel.label;
import static org.neo4j.graphdb.DynamicRelationshipType.withName;

/**
 * Created by william@autoletics.com on 01/03/2015.
 */
public final class InterceptorFactory extends Thread
    implements ProbesInterceptorFactory {

  static final Label THREAD           = label("Thread");
  static final Label FRAME            = label("Frame");
  static final Label NAME             = label("Name");

  static final RelationshipType PROBE = withName("PROBE");
  static final RelationshipType NEXT  = withName("NEXT");
  static final RelationshipType FIRST = withName("FIRST");
  static final RelationshipType LAST  = withName("LAST");

  static final Probes.Name CLOCK_TIME = name("clock").name("time");

  private Options options;

  @Override
  public void init(Probes.Environment env) {

    final GraphDatabaseService db = new GraphDatabaseFactory()
        .newEmbeddedDatabaseBuilder(
            env.getString(name("path")))
        .loadPropertiesFromFile(
            env.getString(name("conf")))
        .newGraphDatabase();

    options = new Options(
        db,
        env.getInt(
            name("batch"),
            1000),
        env.getBoolean(
            name("index"),
            true));

    Runtime.getRuntime().addShutdownHook(this);

  }

  @Override
  public ProbesInterceptor create(Probes.Context ctx) {
    // return a new interceptor instance
    // for each thread metering context
    return new Interceptor(options);
  }

  @Override
  public void run() {

    final Options o = options;

    final GraphDatabaseService db = o.store;

    if(o.index) {

      // create the indices following processing
      try (final Transaction t = db.beginTx()) {

        final Schema s = db.schema();

        // create index Name:name
        s.indexFor(NAME)
            .on("name")
            .create();

        // create index Thread:name
        s.indexFor(THREAD)
            .on("name")
            .create();

        // create index Frame:time
        s.indexFor(FRAME)
            .on("time")
            .create();

        t.success();

      }

    }

    db.shutdown();

  }

  private static final class Interceptor implements ProbesInterceptor {

    // db store options
    final Options options;

    // thread node
    Node thread;

    // a frame stack
    Frame stack;

    // a frame pool
    Frame frames;

    // the current transaction
    // associated with thread
    Transaction transaction;

    // the no of frames
    // within tx window
    int count;

    // a thread specific name-to-node mapping
    final HashMap<Probes.Name, Node> names = new HashMap<>();

    // holds a ref to
    // last root frame
    Node last;

    private Interceptor(Options options) {
      this.options = options;
    }

    private Node thread(GraphDatabaseService db) {

      // create a thread for the context
      final Node tn = db.createNode(THREAD);

      tn.setProperty(
          "name",
          currentThread().getName());

      return tn;

    }

    private Node name(GraphDatabaseService db, Probes.Name n) {

      Node nn;

      // name nodes will be repeated across threads
      // so we will need to merge at end of run
      // (if possible with the neo4j database api)
      if((nn = names.get(n)) == null) {

        // create a thread for the context
        nn = db.createNode(NAME);

        // set the path name for the probe (fqn of method)
        // note: Probes.toString returns the cached path
        nn.setProperty(
            "name",
            n.toString());

        // put in thread specific
        // name->node map instance
        names.put(n, nn);

      }

      return nn;

    }

    private Node frame(GraphDatabaseService db, Probes.Probe p) {

      // create a frame (trace)
      // for current call (probe)
      final Node fn = db.createNode(FRAME);

      // associate the frame with the probe name
      // saving on space for repeating frame names
      fn.createRelationshipTo(
          name(db, p.getName()),
          PROBE);

      // track on frame
      // creation only
      count++;

      return fn;

    }

    /**
     * Called within a real (or simulated) thread
     * when a method (probe) is entered (begun).
     *
     * @param p the probe
     */
    @Override
    public void begin(Probes.Probe p) {

      final Options o = options;

      final Frame s = stack;
      
      final GraphDatabaseService db = o.store;

      Frame f;

      final Node fn;

      // are we at the top (root) of the stack
      if(s == null) {

        // start new transaction batch
        transaction = db.beginTx();

        f = frames;

        Node tn = thread;

        // have we create any frames before
        // for this particular thread (context)
        if(f == null) {

          // initialize the frame pool
          frames = f = new Frame(null);

          // create the the thread node
          // to be associated with roots
          thread = tn = thread(db);

        }

        fn = frame(db, p);

        final Node ln = last;

        // check whether this is the
        // first root frame for thread
        if(ln == null) {

          // associate the thread with the frame (root)
          tn.createRelationshipTo(
              fn,
              FIRST);

        } else {

          // link last root frame
          // to this new root frame
          ln.createRelationshipTo(
              fn,
              NEXT);

        }

        last = fn;

      } else {

        if(count > o.batch) {

          final Transaction t = transaction;

          // commit previous transaction window
          t.success();
          t.close();

          // reset count
          count = 0;

          // start new transaction window
          transaction = db.beginTx();

        }

        if((f = s.next) == null)
          // extend the frame stack
          // we don't trim the pool
          s.next = f = new Frame(s);

        // create a frame (trace) for
        // the current call (probe)
        fn = frame(db, p);

        // the last callee within
        // the current frame scope
        final Node ln = s.last;

        // have we already seen a
        // callee for this frame
        if(ln != null)

          ln.createRelationshipTo(
              fn,
              NEXT);

        else

          s.node.createRelationshipTo(
              fn,
              FIRST);

        s.last = fn;

      }

      // need to track last callee
      f.node = fn;

      // push onto frame stack
      stack = f;

    }

    /**
     * Called within a real (or simulated) thread
     * when a method (probe) is exited (ended).
     *
     * @param p the probe
     */
    @Override
    public void end(Probes.Probe p) {

      final Frame f = stack;

      final Node fn = f.node;

      final Node ln = f.last;

      if(ln != null) {

        // create a relationship
        // with the last frame
        fn.createRelationshipTo(
            ln,
            LAST);

        // need to reset the
        // callee tracking
        f.last = null;

      }

      // add 'time' property to 'Frame' node
      fn.setProperty("time",
          p.reading(CLOCK_TIME).getDelta());

      // we need to commit the transaction
      // when the root call (probe) is
      // pushed off the call (probe) stack
      if((stack = f.prev) == null) {

        final Transaction t = transaction;

        // commit previous transaction window
        t.success();
        t.close();

        // reset counter
        count = 0;

      }

    }

  }

  /**
   * A linked list structure used to track the call stack
   */
  private static final class Frame {

    // the caller frame
    final Frame prev;
    // the callee frame
    Frame next;

    // the node currently associated
    Node node;
    // the last callee (child) node
    Node last;

    private Frame(Frame prev) { this.prev = prev; }

  }

  /**
   * Some config/state options during processing
   */
  private static final class Options {

    final GraphDatabaseService  store;
    final int                   batch;
    final boolean               index;

    public Options(GraphDatabaseService store,
                   int batch,
                   boolean index) {

      this.store = store;
      this.batch = batch;
      this.index = index;

    }

  }

}

jxinsight.override.config

j.s.p.interceptor.enabled=true

# note: this uses version 2.2.3 of the Autoletics performance monitoring (APM) suite
# which supports the use of a wildcard (*) in the definition of interceptor classpath

j.s.p.interceptors=neo4j

j.s.p.interceptor.neo4j.factory.class=com.jinspired.jxinsight.probes.neo4j.gtrace.InterceptorFactory
j.s.p.interceptor.neo4j.factory.classpath=/Users/wlouth/IdeaProjects/neo4j-probes-intg/classes/:/Applications/neo4j-2.2.0/system/lib/*:/Applications/neo4j-2.2.0/lib/*

j.s.p.interceptor.neo4j.environment.imports=path,conf,batch

j.s.p.interceptor.neo4j.environment.import.path=/Applications/neo4j-2.2.0/data/gtrace.db
j.s.p.interceptor.neo4j.environment.import.path.type=string

j.s.p.interceptor.neo4j.environment.import.conf=/Applications/neo4j-2.2.0/conf/neo4j.properties
j.s.p.interceptor.neo4j.environment.import.conf.type=string

j.s.p.interceptor.neo4j.environment.import.batch=5000
j.s.p.interceptor.neo4j.environment.import.batch.type=int