Agent builder exporter

.buildkite/scripts/steps/security/third_party_packages.txt

@@ -50,3 +50,4 @@ patch-package
 @opentelemetry/resources
 @opentelemetry/exporter-logs-otlp-grpc
 @opentelemetry/exporter-logs-otlp-proto
+@opentelemetry/otlp-transformer

package.json

@@ -1313,6 +1313,7 @@
     "@opentelemetry/instrumentation-http": "0.215.0",
     "@opentelemetry/instrumentation-undici": "0.25.0",
     "@opentelemetry/otlp-exporter-base": "0.215.0",
+    "@opentelemetry/otlp-transformer": "0.215.0",
     "@opentelemetry/resources": "2.7.0",
     "@opentelemetry/sdk-logs": "0.215.0",
     "@opentelemetry/semantic-conventions": "1.40.0",

renovate.json

@@ -2336,6 +2336,7 @@
         "@opentelemetry/instrumentation-http",
         "@opentelemetry/instrumentation-undici",
         "@opentelemetry/otlp-exporter-base",
+        "@opentelemetry/otlp-transformer",
         "@opentelemetry/resources",
         "@opentelemetry/sdk-logs",
         "@opentelemetry/semantic-conventions"

src/platform/packages/shared/kbn-tracing/index.ts

@@ -10,3 +10,4 @@
 export { LateBindingSpanProcessor } from './src/late_binding_span_processor';
 export { initTracing } from './src/init_tracing';
 export { OTLPSpanProcessor } from './src/otlp_span_processor';
+export { ElasticsearchOtlpExporter } from './src/elasticsearch_otlp_exporter';

src/platform/packages/shared/kbn-tracing/moon.yml

@@ -21,6 +21,7 @@ dependsOn:
   - '@kbn/std'
   - '@kbn/tracing-config'
   - '@kbn/cleanup-before-exit'
+  - '@kbn/core-elasticsearch-server'
 tags:
   - shared-server
   - package

src/platform/packages/shared/kbn-tracing/src/elasticsearch_otlp_exporter.test.ts

@@ -0,0 +1,186 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import { core } from '@elastic/opentelemetry-node/sdk';
+import type { tracing } from '@elastic/opentelemetry-node/sdk';
+import { ProtobufTraceSerializer } from '@opentelemetry/otlp-transformer';
+import { ElasticsearchOtlpExporter } from './elasticsearch_otlp_exporter';
+import type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+
+jest.mock('@opentelemetry/otlp-transformer', () => ({
+  ProtobufTraceSerializer: {
+    serializeRequest: jest.fn(),
+  },
+}));
+
+const mockedSerializeRequest = ProtobufTraceSerializer.serializeRequest as jest.MockedFunction<
+  typeof ProtobufTraceSerializer.serializeRequest
+>;
+
+describe('ElasticsearchOtlpExporter', () => {
+  const spans = [] as tracing.ReadableSpan[];
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('calls transport.request with correct path, method, and headers on success', (done) => {
+    const serialized = new Uint8Array([1, 2, 3]);
+    mockedSerializeRequest.mockReturnValue(serialized);
+
+    const request = jest.fn().mockResolvedValue({});
+    const client = {
+      transport: { request },
+    } as unknown as ElasticsearchClient;
+
+    const exporter = new ElasticsearchOtlpExporter(client);
+
+    exporter.export(spans, (result) => {
+      try {
+        expect(result.code).toBe(core.ExportResultCode.SUCCESS);
+        expect(request).toHaveBeenCalledWith(
+          {
+            method: 'POST',
+            path: '/_otlp/v1/traces',
+            body: Buffer.from(serialized),
+          },
+          {
+            headers: { 'Content-Type': 'application/x-protobuf' },
+            maxRetries: 3,
+          }
+        );
+        done();
+      } catch (err) {
+        done(err);
+      }
+    });
+  });
+
+  it('returns FAILED when serialization fails', (done) => {
+    mockedSerializeRequest.mockImplementation(() => undefined);
+
+    const request = jest.fn();
+    const client = {
+      transport: { request },
+    } as unknown as ElasticsearchClient;
+
+    const exporter = new ElasticsearchOtlpExporter(client);
+
+    exporter.export(spans, (result) => {
+      try {
+        expect(result.code).toBe(core.ExportResultCode.FAILED);
+        expect(result.error).toEqual(new Error('Serialization failed'));
+        expect(request).not.toHaveBeenCalled();
+        done();
+      } catch (err) {
+        done(err);
+      }
+    });
+  });
+
+  it('returns FAILED when transport.request rejects', (done) => {
+    const serialized = new Uint8Array([9]);
+    mockedSerializeRequest.mockReturnValue(serialized);
+
+    const transportError = new Error('connection reset');
+    const request = jest.fn().mockRejectedValue(transportError);
+    const client = {
+      transport: { request },
+    } as unknown as ElasticsearchClient;
+
+    const exporter = new ElasticsearchOtlpExporter(client);
+
+    exporter.export(spans, (result) => {
+      try {
+        expect(result.code).toBe(core.ExportResultCode.FAILED);
+        expect(result.error).toBe(transportError);
+        done();
+      } catch (err) {
+        done(err);
+      }
+    });
+  });
+
+  it('shutdown waits for in-flight exports to complete before returning', async () => {
+    const serialized = new Uint8Array([1, 2, 3]);
+    mockedSerializeRequest.mockReturnValue(serialized);
+
+    const request = jest.fn().mockResolvedValue({});
+    const client = {
+      transport: { request },
+    } as unknown as ElasticsearchClient;
+
+    const exporter = new ElasticsearchOtlpExporter(client);
+    const results: core.ExportResult[] = [];
+    exporter.export(spans, (result) => results.push(result));
+
+    await exporter.shutdown();
+
+    expect(results).toHaveLength(1);
+    expect(results[0].code).toBe(core.ExportResultCode.SUCCESS);
+  });
+
+  it('export after shutdown returns FAILED immediately', (done) => {
+    const client = {
+      transport: { request: jest.fn() },
+    } as unknown as ElasticsearchClient;
+    const exporter = new ElasticsearchOtlpExporter(client);
+
+    exporter.shutdown().then(() => {
+      exporter.export(spans, (result) => {
+        try {
+          expect(result.code).toBe(core.ExportResultCode.FAILED);
+          expect(result.error?.message).toBe('Exporter has been shut down');
+          expect(client.transport.request).not.toHaveBeenCalled();
+          done();
+        } catch (err) {
+          done(err);
+        }
+      });
+    });
+  });
+
+  it('forceFlush waits for in-flight exports', async () => {
+    const serialized = new Uint8Array([4, 5]);
+    mockedSerializeRequest.mockReturnValue(serialized);
+
+    let resolveRequest: () => void;
+    const request = jest.fn().mockImplementation(
+      () =>
+        new Promise<void>((resolve) => {
+          resolveRequest = resolve;
+        })
+    );
+    const client = {
+      transport: { request },
+    } as unknown as ElasticsearchClient;
+
+    const exporter = new ElasticsearchOtlpExporter(client);
+    const results: core.ExportResult[] = [];
+    exporter.export(spans, (result) => results.push(result));
+
+    const flushPromise = exporter.forceFlush();
+
+    expect(results).toHaveLength(0);
+    resolveRequest!();
+    await flushPromise;
+
+    expect(results).toHaveLength(1);
+    expect(results[0].code).toBe(core.ExportResultCode.SUCCESS);
+  });
+
+  it('forceFlush resolves immediately when no exports are pending', async () => {
+    const client = {
+      transport: { request: jest.fn() },
+    } as unknown as ElasticsearchClient;
+    const exporter = new ElasticsearchOtlpExporter(client);
+
+    await expect(exporter.forceFlush()).resolves.toBeUndefined();
+  });
+});

src/platform/packages/shared/kbn-tracing/src/elasticsearch_otlp_exporter.ts

@@ -0,0 +1,75 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import type { tracing } from '@elastic/opentelemetry-node/sdk';
+import { core } from '@elastic/opentelemetry-node/sdk';
+import { ProtobufTraceSerializer } from '@opentelemetry/otlp-transformer';
+import type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+
+const ES_OTLP_TRACES_PATH = '/_otlp/v1/traces';
+const CONTENT_TYPE_PROTOBUF = 'application/x-protobuf';
+
+/**
+ * A {@link tracing.SpanExporter} that ships OTLP-protobuf encoded spans
+ * to Elasticsearch's native `/_otlp/v1/traces` endpoint via the
+ * ES client transport. This reuses the same connection, auth, and TLS
+ * settings that Kibana already has for talking to Elasticsearch.
+ */
+export class ElasticsearchOtlpExporter implements tracing.SpanExporter {
+  private readonly sendingPromises = new Set<Promise<void>>();
+  private isShutdown = false;
+
+  constructor(private readonly client: ElasticsearchClient) {}
+
+  export(spans: tracing.ReadableSpan[], resultCallback: (result: core.ExportResult) => void): void {
+    if (this.isShutdown) {
+      resultCallback({
+        code: core.ExportResultCode.FAILED,
+        error: new Error('Exporter has been shut down'),
+      });
+      return;
+    }
+
+    const serialized = ProtobufTraceSerializer.serializeRequest(spans);
+    if (!serialized) {
+      resultCallback({
+        code: core.ExportResultCode.FAILED,
+        error: new Error('Serialization failed'),
+      });
+      return;
+    }
+
+    const exportPromise = this.client.transport
+      .request(
+        {
+          method: 'POST',
+          path: ES_OTLP_TRACES_PATH,
+          body: Buffer.from(serialized),
+        },
+        {
+          headers: { 'Content-Type': CONTENT_TYPE_PROTOBUF },
+          maxRetries: 3,
+        }
+      )
+      .then(() => resultCallback({ code: core.ExportResultCode.SUCCESS }))
+      .catch((error) => resultCallback({ code: core.ExportResultCode.FAILED, error }))
+      .finally(() => this.sendingPromises.delete(exportPromise));
+
+    this.sendingPromises.add(exportPromise);
+  }
+
+  async forceFlush(): Promise<void> {
+    await Promise.all(this.sendingPromises);
+  }
+
+  async shutdown(): Promise<void> {
+    this.isShutdown = true;
+    await this.forceFlush();
+  }
+}