feat(opentelemetry-sdk-node): set resources and log provider for experimental start

experimental/CHANGELOG.md

@@ -22,6 +22,7 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 * feat(opentelemetry-sdk-node): set instrumentation and propagators for experimental start [#6148](https://github.com/open-telemetry/opentelemetry-js/pull/6148) @maryliag
 * refactor(configuration): set console exporter as empty object [#6164](https://github.com/open-telemetry/opentelemetry-js/pull/6164) @maryliag
 * feat(instrumentation-http, instrumentation-fetch, instrumentation-xml-http-request): support "QUERY" as a known HTTP method
+* feat(opentelemetry-sdk-node): set resources and log provider for experimental start [#6152](https://github.com/open-telemetry/opentelemetry-js/pull/6152) @maryliag
 
 ### :bug: Bug Fixes
 

experimental/packages/opentelemetry-sdk-node/src/start.ts

@@ -15,16 +15,41 @@
  */
 import {
   ConfigFactory,
+  ConfigurationModel,
   createConfigFactory,
 } from '@opentelemetry/configuration';
-import { diag, DiagConsoleLogger } from '@opentelemetry/api';
 import {
+  context,
+  diag,
+  DiagConsoleLogger,
+  propagation,
+} from '@opentelemetry/api';
+import {
+  getInstanceID,
+  getLogRecordProcessorsFromConfiguration,
   getPropagatorFromConfiguration,
-  setupDefaultContextManager,
-  setupPropagator,
+  getResourceDetectorsFromConfiguration,
+  getResourceFromConfiguration,
 } from './utils';
 import { registerInstrumentations } from '@opentelemetry/instrumentation';
-import type { SDKOptions } from './types';
+import type { SDKComponents, SDKOptions } from './types';
+import { LoggerProvider } from '@opentelemetry/sdk-logs';
+import { logs } from '@opentelemetry/api-logs';
+import {
+  defaultResource,
+  detectResources,
+  Resource,
+  ResourceDetectionConfig,
+  ResourceDetector,
+  resourceFromAttributes,
+} from '@opentelemetry/resources';
+import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
+import { ATTR_SERVICE_INSTANCE_ID } from './semconv';
+import {
+  CompositePropagator,
+  W3CBaggagePropagator,
+  W3CTraceContextPropagator,
+} from '@opentelemetry/core';
 
 /**
  * @experimental Function to start the OpenTelemetry Node SDK
@@ -43,24 +68,104 @@ export function startNodeSDK(sdkOptions: SDKOptions): {
   if (config.log_level != null) {
     diag.setLogger(new DiagConsoleLogger(), { logLevel: config.log_level });
   }
-
   registerInstrumentations({
     instrumentations: sdkOptions?.instrumentations?.flat() ?? [],
   });
-  setupDefaultContextManager();
-  setupPropagator(
-    sdkOptions?.textMapPropagator === null
-      ? null // null means don't set.
-      : (sdkOptions?.textMapPropagator ??
-          getPropagatorFromConfiguration(config))
-  );
+
+  const components = create(config, sdkOptions);
+  context.setGlobalContextManager(components.contextManager);
+  if (components.loggerProvider) {
+    logs.setGlobalLoggerProvider(components.loggerProvider);
+  }
+  if (components.propagator) {
+    propagation.setGlobalPropagator(components.propagator);
+  }
 
   const shutdownFn = async () => {
     const promises: Promise<unknown>[] = [];
+    if (components.loggerProvider) {
+      promises.push(components.loggerProvider.shutdown());
+    }
     await Promise.all(promises);
   };
   return { shutdown: shutdownFn };
 }
 const NOOP_SDK = {
   shutdown: async () => {},
 };
+
+/**
+ * Interpret configuration model and return SDK components.
+ */
+function create(
+  config: ConfigurationModel,
+  sdkOptions: SDKOptions
+): SDKComponents {
+  const defaultContextManager = new AsyncLocalStorageContextManager();
+  defaultContextManager.enable();
+  const components: SDKComponents = {
+    contextManager: defaultContextManager,
+  };
+  const resource = setupResource(config, sdkOptions);
+
+  const propagator =
+    sdkOptions?.textMapPropagator === null
+      ? null // null means don't set.
+      : (sdkOptions?.textMapPropagator ??
+        getPropagatorFromConfiguration(config));
+  if (propagator) {
+    components.propagator = propagator;
+  } else if (propagator === undefined) {
+    components.propagator = new CompositePropagator({
+      propagators: [
+        new W3CTraceContextPropagator(),
+        new W3CBaggagePropagator(),
+      ],
+    });
+  }
+
+  const logProcessors = getLogRecordProcessorsFromConfiguration(config);
+  if (logProcessors) {
+    const loggerProvider = new LoggerProvider({
+      resource: resource,
+      processors: logProcessors,
+    });
+    components.loggerProvider = loggerProvider;
+  }
+
+  return components;
+}
+
+export function setupResource(
+  config: ConfigurationModel,
+  sdkOptions: SDKOptions
+): Resource {
+  let resource: Resource =
+    getResourceFromConfiguration(config) ?? defaultResource();
+  let resourceDetectors: ResourceDetector[] = [];
+
+  if (sdkOptions.resourceDetectors != null) {
+    resourceDetectors = sdkOptions.resourceDetectors;
+  } else if (config.node_resource_detectors) {
+    resourceDetectors = getResourceDetectorsFromConfiguration(config);
+  }
+
+  if (resourceDetectors.length > 0) {
+    const internalConfig: ResourceDetectionConfig = {
+      detectors: resourceDetectors,
+    };
+    resource = resource.merge(detectResources(internalConfig));
+  }
+
+  const instanceId = getInstanceID(config);
+  resource =
+    instanceId === undefined
+      ? resource
+      : resource.merge(
+          resourceFromAttributes({
+            [ATTR_SERVICE_INSTANCE_ID]: instanceId,
+          })
+        );
+
+  return resource;
+}

experimental/packages/opentelemetry-sdk-node/src/types.ts

@@ -18,15 +18,20 @@ import type { ContextManager } from '@opentelemetry/api';
 import { TextMapPropagator } from '@opentelemetry/api';
 import { Instrumentation } from '@opentelemetry/instrumentation';
 import { Resource, ResourceDetector } from '@opentelemetry/resources';
-import { LogRecordProcessor } from '@opentelemetry/sdk-logs';
-import { IMetricReader, ViewOptions } from '@opentelemetry/sdk-metrics';
+import { LoggerProvider, LogRecordProcessor } from '@opentelemetry/sdk-logs';
+import {
+  IMetricReader,
+  MeterProvider,
+  ViewOptions,
+} from '@opentelemetry/sdk-metrics';
 import {
   Sampler,
   SpanExporter,
   SpanLimits,
   SpanProcessor,
   IdGenerator,
 } from '@opentelemetry/sdk-trace-base';
+import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
 
 export interface NodeSDKConfiguration {
   autoDetectResources: boolean;
@@ -56,5 +61,14 @@ export interface NodeSDKConfiguration {
  */
 export interface SDKOptions {
   instrumentations?: (Instrumentation | Instrumentation[])[];
+  resourceDetectors?: ResourceDetector[];
   textMapPropagator?: TextMapPropagator | null;
 }
+
+export interface SDKComponents {
+  contextManager: ContextManager;
+  loggerProvider?: LoggerProvider;
+  meterProvider?: MeterProvider;
+  tracesProvider?: NodeTracerProvider;
+  propagator?: TextMapPropagator;
+}

experimental/packages/opentelemetry-sdk-node/src/utils.ts

@@ -33,11 +33,14 @@ import { OTLPTraceExporter as OTLPHttpTraceExporter } from '@opentelemetry/expor
 import { OTLPTraceExporter as OTLPGrpcTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc';
 import { ZipkinExporter } from '@opentelemetry/exporter-zipkin';
 import {
+  DetectedResourceAttributes,
   envDetector,
   hostDetector,
   osDetector,
   processDetector,
+  Resource,
   ResourceDetector,
+  resourceFromAttributes,
   serviceInstanceIdDetector,
 } from '@opentelemetry/resources';
 import {
@@ -50,14 +53,44 @@ import {
 import { B3InjectEncoding, B3Propagator } from '@opentelemetry/propagator-b3';
 import { JaegerPropagator } from '@opentelemetry/propagator-jaeger';
 import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
-import { ConfigurationModel } from '@opentelemetry/configuration';
+import {
+  BatchLogRecordProcessor,
+  ConsoleLogRecordExporter,
+  LogRecordExporter,
+  LogRecordProcessor,
+  SimpleLogRecordProcessor,
+} from '@opentelemetry/sdk-logs';
+import { OTLPLogExporter as OTLPHttpLogExporter } from '@opentelemetry/exporter-logs-otlp-http';
+import { OTLPLogExporter as OTLPGrpcLogExporter } from '@opentelemetry/exporter-logs-otlp-grpc';
+import { OTLPLogExporter as OTLPProtoLogExporter } from '@opentelemetry/exporter-logs-otlp-proto';
+import { CompressionAlgorithm } from '@opentelemetry/otlp-exporter-base';
+import {
+  ConfigurationModel,
+  LogRecordExporterModel,
+} from '@opentelemetry/configuration';
 
 const RESOURCE_DETECTOR_ENVIRONMENT = 'env';
 const RESOURCE_DETECTOR_HOST = 'host';
 const RESOURCE_DETECTOR_OS = 'os';
 const RESOURCE_DETECTOR_PROCESS = 'process';
 const RESOURCE_DETECTOR_SERVICE_INSTANCE_ID = 'serviceinstance';
 
+export function getResourceFromConfiguration(
+  config: ConfigurationModel
+): Resource | undefined {
+  if (config.resource && config.resource.attributes) {
+    const attr: DetectedResourceAttributes = {};
+    for (let i = 0; i < config.resource.attributes.length; i++) {
+      const a = config.resource.attributes[i];
+      attr[a.name] = a.value;
+    }
+    return resourceFromAttributes(attr, {
+      schemaUrl: config.resource.schema_url,
+    });
+  }
+  return undefined;
+}
+
 export function getResourceDetectorsFromEnv(): Array<ResourceDetector> {
   // When updating this list, make sure to also update the section `resourceDetectors` on README.
   const resourceDetectors = new Map<string, ResourceDetector>([
@@ -91,6 +124,37 @@ export function getResourceDetectorsFromEnv(): Array<ResourceDetector> {
   });
 }
 
+export function getResourceDetectorsFromConfiguration(
+  config: ConfigurationModel
+): Array<ResourceDetector> {
+  // When updating this list, make sure to also update the section `resourceDetectors` on README.
+  const resourceDetectors = new Map<string, ResourceDetector>([
+    [RESOURCE_DETECTOR_ENVIRONMENT, envDetector],
+    [RESOURCE_DETECTOR_HOST, hostDetector],
+    [RESOURCE_DETECTOR_OS, osDetector],
+    [RESOURCE_DETECTOR_SERVICE_INSTANCE_ID, serviceInstanceIdDetector],
+    [RESOURCE_DETECTOR_PROCESS, processDetector],
+  ]);
+
+  const resourceDetectorsFromConfig = config.node_resource_detectors ?? [];
+
+  if (resourceDetectorsFromConfig.includes('all')) {
+    return [...resourceDetectors.values()].flat();
+  }
+
+  if (resourceDetectorsFromConfig.includes('none')) {
+    return [];
+  }
+
+  return resourceDetectorsFromConfig.flatMap(detector => {
+    const resourceDetector = resourceDetectors.get(detector);
+    if (!resourceDetector) {
+      diag.warn(`Invalid resource detector "${detector}" specified`);
+    }
+    return resourceDetector || [];
+  });
+}
+
 export function getOtlpProtocolFromEnv(): string {
   return (
     getStringFromEnv('OTEL_EXPORTER_OTLP_TRACES_PROTOCOL') ??
@@ -310,12 +374,6 @@ export function setupContextManager(
   context.setGlobalContextManager(contextManager);
 }
 
-export function setupDefaultContextManager() {
-  const defaultContextManager = new AsyncLocalStorageContextManager();
-  defaultContextManager.enable();
-  context.setGlobalContextManager(defaultContextManager);
-}
-
 export function setupPropagator(
   propagator: TextMapPropagator | null | undefined
 ) {
@@ -350,3 +408,76 @@ export function getKeyListFromObjectArray(
     .map(item => Object.keys(item))
     .reduce((prev, curr) => prev.concat(curr), []);
 }
+
+export function getLogRecordExporter(
+  exporter: LogRecordExporterModel
+): LogRecordExporter {
+  if (exporter.otlp_http) {
+    const encoding = exporter.otlp_http.encoding;
+    if (encoding === 'json') {
+      return new OTLPHttpLogExporter({
+        compression:
+          exporter.otlp_http.compression === 'gzip'
+            ? CompressionAlgorithm.GZIP
+            : CompressionAlgorithm.NONE,
+      });
+    }
+    if (encoding === 'protobuf') {
+      return new OTLPProtoLogExporter();
+    }
+    diag.warn(
+      `Unsupported OTLP logs encoding: ${encoding}. Using http/protobuf.`
+    );
+    return new OTLPProtoLogExporter();
+  } else if (exporter.otlp_grpc) {
+    return new OTLPGrpcLogExporter();
+  } else if (exporter.console) {
+    return new ConsoleLogRecordExporter();
+  }
+  diag.warn(`Unsupported Exporter value. Using OTLP http/protobuf.`);
+  return new OTLPProtoLogExporter();
+}
+
+export function getLogRecordProcessorsFromConfiguration(
+  config: ConfigurationModel
+): LogRecordProcessor[] | undefined {
+  const logRecordProcessors: LogRecordProcessor[] = [];
+  config.logger_provider?.processors?.forEach(processor => {
+    if (processor.batch) {
+      logRecordProcessors.push(
+        new BatchLogRecordProcessor(
+          getLogRecordExporter(processor.batch.exporter),
+          {
+            maxQueueSize: processor.batch.max_queue_size,
+            maxExportBatchSize: processor.batch.max_export_batch_size,
+            scheduledDelayMillis: processor.batch.schedule_delay,
+            exportTimeoutMillis: processor.batch.export_timeout,
+          }
+        )
+      );
+    }
+    if (processor.simple) {
+      logRecordProcessors.push(
+        new SimpleLogRecordProcessor(
+          getLogRecordExporter(processor.simple.exporter)
+        )
+      );
+    }
+  });
+  if (logRecordProcessors.length > 0) {
+    return logRecordProcessors;
+  }
+  return undefined;
+}
+
+export function getInstanceID(config: ConfigurationModel): string | undefined {
+  if (config.resource?.attributes) {
+    for (let i = 0; i < config.resource.attributes.length; i++) {
+      const element = config.resource.attributes[i];
+      if (element.name === 'service.instance.id') {
+        return element.value?.toString();
+      }
+    }
+  }
+  return undefined;
+}