feat: Implement experimental AWS ECS resource attributes

detectors/node/opentelemetry-resource-detector-aws/src/detectors/AwsEcsDetector.ts

@@ -15,50 +15,68 @@
  */
 
 import { diag } from '@opentelemetry/api';
-import {
-  Detector,
-  Resource,
-  ResourceDetectionConfig,
-} from '@opentelemetry/resources';
+import { Detector, Resource } from '@opentelemetry/resources';
 import {
   CloudProviderValues,
   CloudPlatformValues,
   SemanticResourceAttributes,
 } from '@opentelemetry/semantic-conventions';
+import * as http from 'http';
 import * as util from 'util';
 import * as fs from 'fs';
 import * as os from 'os';
 import { getEnv } from '@opentelemetry/core';
 
+const HTTP_TIMEOUT_IN_MS = 1000;
+
+interface AwsLogOptions {
+  readonly 'awslogs-region'?: string;
+  readonly 'awslogs-group'?: string;
+  readonly 'awslogs-stream'?: string;
+}
+
 /**
  * The AwsEcsDetector can be used to detect if a process is running in AWS
  * ECS and return a {@link Resource} populated with data about the ECS
  * plugins of AWS X-Ray. Returns an empty Resource if detection fails.
  */
 export class AwsEcsDetector implements Detector {
-  readonly CONTAINER_ID_LENGTH = 64;
-  readonly DEFAULT_CGROUP_PATH = '/proc/self/cgroup';
+  static readonly CONTAINER_ID_LENGTH = 64;
+  static readonly DEFAULT_CGROUP_PATH = '/proc/self/cgroup';
+
   private static readFileAsync = util.promisify(fs.readFile);
 
-  async detect(_config?: ResourceDetectionConfig): Promise<Resource> {
+  async detect(): Promise<Resource> {
     const env = getEnv();
     if (!env.ECS_CONTAINER_METADATA_URI_V4 && !env.ECS_CONTAINER_METADATA_URI) {
       diag.debug('AwsEcsDetector failed: Process is not on ECS');
       return Resource.empty();
     }
 
-    const hostName = os.hostname();
-    const containerId = await this._getContainerId();
-
-    return !hostName && !containerId
-      ? Resource.empty()
-      : new Resource({
-          [SemanticResourceAttributes.CLOUD_PROVIDER]: CloudProviderValues.AWS,
-          [SemanticResourceAttributes.CLOUD_PLATFORM]:
-            CloudPlatformValues.AWS_ECS,
-          [SemanticResourceAttributes.CONTAINER_NAME]: hostName || '',
-          [SemanticResourceAttributes.CONTAINER_ID]: containerId || '',
-        });
+    let resource = new Resource({
+      [SemanticResourceAttributes.CLOUD_PROVIDER]: CloudProviderValues.AWS,
+      [SemanticResourceAttributes.CLOUD_PLATFORM]: CloudPlatformValues.AWS_ECS,
+    }).merge(await AwsEcsDetector._getContainerIdAndHostnameResource());
+
+    const metadataUrl = getEnv().ECS_CONTAINER_METADATA_URI_V4;
+    if (metadataUrl) {
+      const [containerMetadata, taskMetadata] = await Promise.all([
+        AwsEcsDetector._getUrlAsJson(metadataUrl),
+        AwsEcsDetector._getUrlAsJson(`${metadataUrl}/task`),
+      ]);
+
+      const metadatav4Resource = await AwsEcsDetector._getMetadataV4Resource(
+        containerMetadata,
+        taskMetadata
+      );
+      const logsResource = await AwsEcsDetector._getLogResource(
+        containerMetadata
+      );
+
+      resource = resource.merge(metadatav4Resource).merge(logsResource);
+    }
+
+    return resource;
   }
 
   /**
@@ -68,22 +86,146 @@ export class AwsEcsDetector implements Detector {
    * we do not throw an error but throw warning message
    * and then return null string
    */
-  private async _getContainerId(): Promise<string | undefined> {
+  private static async _getContainerIdAndHostnameResource(): Promise<Resource> {
+    const hostName = os.hostname();
+
+    let containerId = '';
     try {
       const rawData = await AwsEcsDetector.readFileAsync(
-        this.DEFAULT_CGROUP_PATH,
+        AwsEcsDetector.DEFAULT_CGROUP_PATH,
         'utf8'
       );
       const splitData = rawData.trim().split('\n');
       for (const str of splitData) {
-        if (str.length > this.CONTAINER_ID_LENGTH) {
-          return str.substring(str.length - this.CONTAINER_ID_LENGTH);
+        if (str.length > AwsEcsDetector.CONTAINER_ID_LENGTH) {
+          containerId = str.substring(
+            str.length - AwsEcsDetector.CONTAINER_ID_LENGTH
+          );
+          break;
         }
       }
     } catch (e) {
-      diag.warn(`AwsEcsDetector failed to read container ID: ${e.message}`);
+      diag.warn('AwsEcsDetector failed to read container ID', e);
+    }
+
+    if (hostName || containerId) {
+      return new Resource({
+        [SemanticResourceAttributes.CONTAINER_NAME]: hostName || '',
+        [SemanticResourceAttributes.CONTAINER_ID]: containerId || '',
+      });
+    }
+
+    return Resource.empty();
+  }
+
+  private static async _getMetadataV4Resource(
+    containerMetadata: any,
+    taskMetadata: any
+  ): Promise<Resource> {
+    const launchType: string = taskMetadata['LaunchType'];
+    const taskArn: string = taskMetadata['TaskARN'];
+
+    const baseArn: string = taskArn.substring(0, taskArn.lastIndexOf(':'));
+    const cluster: string = taskMetadata['Cluster'];
+
+    const clusterArn = cluster.startsWith('arn:')
+      ? cluster
+      : `${baseArn}:cluster/${cluster}`;
+
+    const containerArn: string = containerMetadata['ContainerARN'];
+
+    // https://github.com/open-telemetry/opentelemetry-specification/blob/main/semantic_conventions/resource/cloud_provider/aws/ecs.yaml
+    return new Resource({
+      [SemanticResourceAttributes.AWS_ECS_CONTAINER_ARN]: containerArn,
+      [SemanticResourceAttributes.AWS_ECS_CLUSTER_ARN]: clusterArn,
+      [SemanticResourceAttributes.AWS_ECS_LAUNCHTYPE]:
+        launchType?.toLowerCase(),
+      [SemanticResourceAttributes.AWS_ECS_TASK_ARN]: taskArn,
+      [SemanticResourceAttributes.AWS_ECS_TASK_FAMILY]: taskMetadata['Family'],
+      [SemanticResourceAttributes.AWS_ECS_TASK_REVISION]:
+        taskMetadata['Revision'],
+    });
+  }
+
+  private static async _getLogResource(
+    containerMetadata: any
+  ): Promise<Resource> {
+    if (
+      containerMetadata['LogDriver'] !== 'awslogs' ||
+      !containerMetadata['LogOptions']
+    ) {
+      return Resource.EMPTY;
     }
-    return undefined;
+
+    const containerArn = containerMetadata['ContainerARN']!;
+    const logOptions = containerMetadata['LogOptions'] as AwsLogOptions;
+
+    const logsRegion =
+      logOptions['awslogs-region'] ||
+      AwsEcsDetector._getRegionFromArn(containerArn);
+
+    const awsAccount = AwsEcsDetector._getAccountFromArn(containerArn);
+
+    const logsGroupName = logOptions['awslogs-group']!;
+    const logsGroupArn = `arn:aws:logs:${logsRegion}:${awsAccount}:log-group:${logsGroupName}`;
+    const logsStreamName = logOptions['awslogs-stream']!;
+    const logsStreamArn = `arn:aws:logs:${logsRegion}:${awsAccount}:log-group:${logsGroupName}:log-stream:${logsStreamName}`;
+
+    return new Resource({
+      [SemanticResourceAttributes.AWS_LOG_GROUP_NAMES]: [logsGroupName],
+      [SemanticResourceAttributes.AWS_LOG_GROUP_ARNS]: [logsGroupArn],
+      [SemanticResourceAttributes.AWS_LOG_STREAM_NAMES]: [logsStreamName],
+      [SemanticResourceAttributes.AWS_LOG_STREAM_ARNS]: [logsStreamArn],
+    });
+  }
+
+  private static _getAccountFromArn(containerArn: string): string {
+    const match = /arn:aws:ecs:[^:]+:([^:]+):.*/.exec(containerArn);
+    return match![1];
+  }
+
+  private static _getRegionFromArn(containerArn: string): string {
+    const match = /arn:aws:ecs:([^:]+):.*/.exec(containerArn);
+    return match![1];
+  }
+
+  private static _getUrlAsJson(url: string): Promise<any> {
+    return new Promise<string>((resolve, reject) => {
+      const request = http.get(url, (response: http.IncomingMessage) => {
+        if (response.statusCode && response.statusCode >= 400) {
+          reject(
+            new Error(
+              `Request to '${url}' failed with status ${response.statusCode}`
+            )
+          );
+        }
+        /*
+         * Concatenate the response out of chunks:
+         * https://nodejs.org/api/stream.html#stream_event_data
+         */
+        let responseBody = '';
+        response.on(
+          'data',
+          (chunk: Buffer) => (responseBody += chunk.toString())
+        );
+        // All the data has been read, resolve the Promise
+        response.on('end', () => resolve(responseBody));
+        /*
+         * https://nodejs.org/api/http.html#httprequesturl-options-callback, see the
+         * 'In the case of a premature connection close after the response is received'
+         * case
+         */
+        request.on('error', reject);
+      });
+
+      // Set an aggressive timeout to prevent lock-ups
+      request.setTimeout(HTTP_TIMEOUT_IN_MS, () => {
+        request.destroy();
+      });
+      // Connection error, disconnection, etc.
+      request.on('error', reject);
+      request.end();
+    }).then(responseBodyRaw => JSON.parse(responseBodyRaw));
   }
 }