[APM] Generate breakdown metrics

packages/elastic-apm-generator/src/index.ts

@@ -12,3 +12,4 @@ export { getTransactionMetrics } from './lib/utils/get_transaction_metrics';
 export { getSpanDestinationMetrics } from './lib/utils/get_span_destination_metrics';
 export { getObserverDefaults } from './lib/defaults/get_observer_defaults';
 export { toElasticsearchOutput } from './lib/output/to_elasticsearch_output';
+export { getBreakdownMetrics } from './lib/utils/get_breakdown_metrics';

packages/elastic-apm-generator/src/lib/base_span.ts

@@ -8,10 +8,12 @@
 
 import { Fields } from './entity';
 import { Serializable } from './serializable';
+import { Span } from './span';
+import { Transaction } from './transaction';
 import { generateTraceId } from './utils/generate_id';
 
 export class BaseSpan extends Serializable {
-  private _children: BaseSpan[] = [];
+  private readonly _children: BaseSpan[] = [];
 
   constructor(fields: Fields) {
     super({
@@ -22,20 +24,29 @@ export class BaseSpan extends Serializable {
     });
   }
 
-  traceId(traceId: string) {
-    this.fields['trace.id'] = traceId;
+  parent(span: BaseSpan) {
+    this.fields['trace.id'] = span.fields['trace.id'];
+    this.fields['parent.id'] = span.isSpan()
+      ? span.fields['span.id']
+      : span.fields['transaction.id'];
+
+    if (this.isSpan()) {
+      this.fields['transaction.id'] = span.fields['transaction.id'];
+    }
     this._children.forEach((child) => {
-      child.fields['trace.id'] = traceId;
+      child.parent(this);
     });
+
     return this;
   }
 
   children(...children: BaseSpan[]) {
-    this._children.push(...children);
     children.forEach((child) => {
-      child.traceId(this.fields['trace.id']!);
+      child.parent(this);
     });
 
+    this._children.push(...children);
+
     return this;
   }
 
@@ -52,4 +63,12 @@ export class BaseSpan extends Serializable {
   serialize(): Fields[] {
     return [this.fields, ...this._children.flatMap((child) => child.serialize())];
   }
+
+  isSpan(): this is Span {
+    return this.fields['processor.event'] === 'span';
+  }
+
+  isTransaction(): this is Transaction {
+    return this.fields['processor.event'] === 'transaction';
+  }
 }

packages/elastic-apm-generator/src/lib/entity.ts

@@ -10,9 +10,11 @@ export type Fields = Partial<{
   '@timestamp': number;
   'agent.name': string;
   'agent.version': string;
+  'container.id': string;
   'ecs.version': string;
   'event.outcome': string;
   'event.ingested': number;
+  'host.name': string;
   'metricset.name': string;
   'observer.version': string;
   'observer.version_major': number;
@@ -42,6 +44,8 @@ export type Fields = Partial<{
   'span.destination.service.type': string;
   'span.destination.service.response_time.sum.us': number;
   'span.destination.service.response_time.count': number;
+  'span.self_time.count': number;
+  'span.self_time.sum.us': number;
 }>;
 
 export class Entity {

packages/elastic-apm-generator/src/lib/output/to_elasticsearch_output.ts

@@ -14,10 +14,12 @@ export function toElasticsearchOutput(events: Fields[], versionOverride?: string
   return events.map((event) => {
     const values = {
       ...event,
+      ...getObserverDefaults(),
       '@timestamp': new Date(event['@timestamp']!).toISOString(),
       'timestamp.us': event['@timestamp']! * 1000,
       'ecs.version': '1.4',
-      ...getObserverDefaults(),
+      'service.node.name':
+        event['service.node.name'] || event['container.id'] || event['host.name'],
     };
 
     const document = {};

packages/elastic-apm-generator/src/lib/service.ts

@@ -14,6 +14,7 @@ export class Service extends Entity {
     return new Instance({
       ...this.fields,
       ['service.node.name']: instanceName,
+      'container.id': instanceName,
     });
   }
 }

packages/elastic-apm-generator/src/lib/span.ts

@@ -19,18 +19,6 @@ export class Span extends BaseSpan {
     });
   }
 
-  children(...children: BaseSpan[]) {
-    super.children(...children);
-
-    children.forEach((child) =>
-      child.defaults({
-        'parent.id': this.fields['span.id'],
-      })
-    );
-
-    return this;
-  }
-
   duration(duration: number) {
     this.fields['span.duration.us'] = duration * 1000;
     return this;

packages/elastic-apm-generator/src/lib/transaction.ts

@@ -11,6 +11,8 @@ import { Fields } from './entity';
 import { generateEventId } from './utils/generate_id';
 
 export class Transaction extends BaseSpan {
+  private _sampled: boolean = true;
+
   constructor(fields: Fields) {
     super({
       ...fields,
@@ -19,19 +21,25 @@ export class Transaction extends BaseSpan {
       'transaction.sampled': true,
     });
   }
-  children(...children: BaseSpan[]) {
-    super.children(...children);
-    children.forEach((child) =>
-      child.defaults({
-        'transaction.id': this.fields['transaction.id'],
-        'parent.id': this.fields['transaction.id'],
-      })
-    );
-    return this;
-  }
 
   duration(duration: number) {
     this.fields['transaction.duration.us'] = duration * 1000;
     return this;
   }
+
+  sample(sampled: boolean = true) {
+    this._sampled = sampled;
+    return this;
+  }
+
+  serialize() {
+    const [transaction, ...spans] = super.serialize();
+
+    const events = [transaction];
+    if (this._sampled) {
+      events.push(...spans);
+    }
+
+    return events;
+  }
 }

packages/elastic-apm-generator/src/lib/utils/aggregate.ts

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+import moment from 'moment';
+import { pickBy } from 'lodash';
+import objectHash from 'object-hash';
+import { Fields } from '../entity';
+import { createPicker } from './create_picker';
+
+export function aggregate(events: Fields[], fields: string[]) {
+  const picker = createPicker(fields);
+
+  const metricsets = new Map<string, { key: Fields; events: Fields[] }>();
+
+  function getMetricsetKey(span: Fields) {
+    const timestamp = moment(span['@timestamp']).valueOf();
+    return {
+      '@timestamp': timestamp - (timestamp % (60 * 1000)),
+      ...pickBy(span, picker),
+    };
+  }
+
+  for (const event of events) {
+    const key = getMetricsetKey(event);
+    const id = objectHash(key);
+
+    let metricset = metricsets.get(id);
+
+    if (!metricset) {
+      metricset = {
+        key: { ...key, 'processor.event': 'metric', 'processor.name': 'metric' },
+        events: [],
+      };
+      metricsets.set(id, metricset);
+    }
+
+    metricset.events.push(event);
+  }
+
+  return Array.from(metricsets.values());
+}

packages/elastic-apm-generator/src/lib/utils/create_picker.ts

@@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+export function createPicker(fields: string[]) {
+  const wildcards = fields
+    .filter((field) => field.endsWith('.*'))
+    .map((field) => field.replace('*', ''));
+
+  return (value: unknown, key: string) => {
+    return fields.includes(key) || wildcards.some((field) => key.startsWith(field));
+  };
+}

packages/elastic-apm-generator/src/lib/utils/get_breakdown_metrics.ts

@@ -0,0 +1,145 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+import objectHash from 'object-hash';
+import { groupBy, pickBy } from 'lodash';
+import { Fields } from '../entity';
+import { createPicker } from './create_picker';
+
+const instanceFields = [
+  'container.*',
+  'kubernetes.*',
+  'agent.*',
+  'process.*',
+  'cloud.*',
+  'service.*',
+  'host.*',
+];
+
+const instancePicker = createPicker(instanceFields);
+
+const metricsetPicker = createPicker([
+  'transaction.type',
+  'transaction.name',
+  'span.type',
+  'span.subtype',
+]);
+
+export function getBreakdownMetrics(events: Fields[]) {
+  const txWithSpans = groupBy(
+    events.filter(
+      (event) => event['processor.event'] === 'span' || event['processor.event'] === 'transaction'
+    ),
+    (event) => event['transaction.id']
+  );
+
+  const metricsets: Map<string, Fields> = new Map();
+
+  Object.keys(txWithSpans).forEach((transactionId) => {
+    const txEvents = txWithSpans[transactionId];
+    const transaction = txEvents.find((event) => event['processor.event'] === 'transaction')!;
+
+    const eventsById: Record<string, Fields> = {};
+    const activityByParentId: Record<string, Array<{ from: number; to: number }>> = {};
+    for (const event of txEvents) {
+      const id =
+        event['processor.event'] === 'transaction' ? event['transaction.id'] : event['span.id'];
+      eventsById[id!] = event;
+
+      const parentId = event['parent.id'];
+
+      if (!parentId) {
+        continue;
+      }
+
+      if (!activityByParentId[parentId]) {
+        activityByParentId[parentId] = [];
+      }
+
+      const from = event['@timestamp']! * 1000;
+      const to =
+        from +
+        (event['processor.event'] === 'transaction'
+          ? event['transaction.duration.us']!
+          : event['span.duration.us']!);
+
+      activityByParentId[parentId].push({ from, to });
+    }
+
+    // eslint-disable-next-line guard-for-in
+    for (const id in eventsById) {
+      const event = eventsById[id];
+      const activities = activityByParentId[id] || [];
+
+      const timeStart = event['@timestamp']! * 1000;
+
+      let selfTime = 0;
+      let lastMeasurement = timeStart;
+      const changeTimestamps = [
+        ...new Set([
+          timeStart,
+          ...activities.flatMap((activity) => [activity.from, activity.to]),
+          timeStart +
+            (event['processor.event'] === 'transaction'
+              ? event['transaction.duration.us']!
+              : event['span.duration.us']!),
+        ]),
+      ];
+
+      for (const timestamp of changeTimestamps) {
+        const hasActiveChildren = activities.some(
+          (activity) => activity.from < timestamp && activity.to >= timestamp
+        );
+
+        if (!hasActiveChildren) {
+          selfTime += timestamp - lastMeasurement;
+        }
+
+        lastMeasurement = timestamp;
+      }
+
+      const key = {
+        '@timestamp': event['@timestamp']! - (event['@timestamp']! % (30 * 1000)),
+        'transaction.type': transaction['transaction.type'],
+        'transaction.name': transaction['transaction.name'],
+        ...pickBy(event, metricsetPicker),
+      };
+
+      const instance = pickBy(event, instancePicker);
+
+      const metricsetId = objectHash(key);
+
+      let metricset = metricsets.get(metricsetId);
+
+      if (!metricset) {
+        metricset = {
+          ...key,
+          ...instance,
+          'processor.event': 'metric',
+          'processor.name': 'metric',
+          'metricset.name': `span_breakdown`,
+          'span.self_time.count': 0,
+          'span.self_time.sum.us': 0,
+        };
+
+        if (event['processor.event'] === 'transaction') {
+          metricset['span.type'] = 'app';
+        } else {
+          metricset['span.type'] = event['span.type'];
+          metricset['span.subtype'] = event['span.subtype'];
+        }
+
+        metricsets.set(metricsetId, metricset);
+      }
+
+      metricset['span.self_time.count']!++;
+      metricset['span.self_time.sum.us']! += selfTime;
+    }
+  });
+
+  return Array.from(metricsets.values());
+}