update url for the new opensearch otel demo repo

update integration script to log into a file
update the README.md for the otel-collector pipline descrition
update ports for the otel-col health-check
add memory_limiter / servicegraph processors for otel-col

Signed-off-by: YANGDB <[email protected]>

.env

@@ -10,6 +10,7 @@ ENV_PLATFORM=local
 # OpenTelemetry Collector
 OTEL_COLLECTOR_HOST=otelcol
 OTEL_COLLECTOR_PORT=4317
+OTEL_COLLECTOR_HEALTH_CHECK_PORT=1313
 OTEL_EXPORTER_OTLP_ENDPOINT=http://${OTEL_COLLECTOR_HOST}:${OTEL_COLLECTOR_PORT}
 PUBLIC_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:4318/v1/traces
 

.github/architecture.md

@@ -37,15 +37,15 @@ Backend supportive services
  - [Frontend Nginx Proxy](http://nginx:90) *(replacement for _Frontend-Proxy_)*
    - See [description](../src/nginx-otel/README.md)
  - [OpenSearch](https://opensearch-node1:9200)
-    - See [description](https://github.com/YANG-DB/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L697)
+    - See [description](https://github.com/opensearch-project/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L697)
  - [Dashboards](http://opensearch-dashboards:5601)
-   - See [description](https://github.com/YANG-DB/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L747) 
+   - See [description](https://github.com/opensearch-project/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L747) 
  - [Prometheus](http://prometheus:9090)
-   - See [description](https://github.com/YANG-DB/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L674)
+   - See [description](https://github.com/opensearch-project/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L674)
  - [Feature-Flag](http://feature-flag-service:8881)
    - See [description](../src/featureflagservice/README.md)
  - [Grafana](http://grafana:3000)
-   - See [description](https://github.com/YANG-DB/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L637)
+   - See [description](https://github.com/opensearch-project/opentelemetry-demo/blob/12d52cbb23bbf4226f6de2dfec840482a0a7d054/docker-compose.yml#L637)
 
 ### Services Topology
 The next diagram shows the docker compose services dependencies

docker-compose.yml

@@ -677,10 +677,11 @@ services:
       - ./src/otelcollector/otelcol-config.yml:/etc/otelcol-config.yml
       - ./src/otelcollector/otelcol-config-extras.yml:/etc/otelcol-config-extras.yml
     ports:
-      - "4317"          # OTLP over gRPC receiver
-      - "4318:4318"     # OTLP over HTTP receiver
-      - "9464"          # Prometheus exporter
-      - "8888"          # metrics endpoint
+      - "4317"            # OTLP over gRPC receiver
+      - "4318:4318"       # OTLP over HTTP receiver
+      - "13133:13133"     # health check port
+      - "9464"            # Prometheus exporter
+      - "8888"            # metrics endpoint
     depends_on:
       - jaeger-agent
       - data-prepper

src/dataprepper/README.md

@@ -50,3 +50,6 @@ There are some additional `trace.group` related fields which are not part of the
 - traceGroupFields.durationInNanos - A derived field, the durationInNanos of the trace's root span.
 
 ```
+### Metrics from Traces Processors
+
+Adding new processors for creating metrics for logs and traces that pass through [Data Prepper](https://opensearch.org/blog/Announcing-Data-Prepper-2.1.0/)

src/integrations/src/install.py

@@ -13,8 +13,13 @@
 # For testing only. Don't store credentials in code.
 auth = ('admin', 'admin')
 
-# Configure logging
-logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s', level=logging.INFO)
+# Configure logging to file
+logging.basicConfig(
+    filename='application.log',  # The file where the logs should be written to
+    filemode='a',  # Append mode, use 'w' for overwriting each time the script is run
+    format='%(asctime)s [%(levelname)s] %(message)s',
+    level=logging.INFO
+)
 logger = logging.getLogger(__name__)
 
 # Create the client with SSL/TLS enabled, but hostname verification disabled.
@@ -28,10 +33,9 @@
     ssl_show_warn = False
 )
 # verify connection to opensearch
-# verify connection to opensearch
 def test_connection():
-    max_retries = 30  # Maximum number of retries
-    retry_interval = 10  # Wait for 10 seconds between retries
+    max_retries = 100  # Maximum number of retries
+    retry_interval = 20  # Wait for 20 seconds between retries
 
     for i in range(max_retries):
         try:
@@ -42,15 +46,15 @@ def test_connection():
                 verify=False  # Disable SSL verification
             )
             response.raise_for_status()  # Raise an exception if the request failed
-            print('Successfully connected to OpenSearch')
+            logger.info('Successfully connected to OpenSearch')
             return  # Exit the function if connection is successful
         except requests.HTTPError as e:
             logging.error(f'Failed to connect to OpenSearch, error: {str(e)}')
 
-        print(f'Attempt {i + 1} failed, waiting for {retry_interval} seconds before retrying...')
+        logger.info(f'Attempt {i + 1} failed, waiting for {retry_interval} seconds before retrying...')
         time.sleep(retry_interval)
 
-    print(f'Failed to connect to OpenSearch after {max_retries} attempts')
+    logger.error(f'Failed to connect to OpenSearch after {max_retries} attempts')
     exit(1)  # Exit the program with an error code
 
 # create mapping components to compose the different observability categories
@@ -62,7 +66,7 @@ def create_mapping_components(client):
                 mapping = json.load(f)
 
                 template_name = os.path.splitext(filename)[0]  # Remove the .mapping extension
-                print(f'About to load  template: {template_name}')
+                logger.info(f'About to load  template: {template_name}')
                 # Create the component template
                 try:
                     response = requests.put(
@@ -74,7 +78,6 @@ def create_mapping_components(client):
                     )
                     response.raise_for_status()  # Raise an exception if the request failed
                     logger.info(f'Successfully created component template: {template_name}')
-                    print(f'Successfully created component template: {template_name}')
                 except requests.HTTPError as e:
                     logger.error(f'Failed to create component template: {template_name}, error: {str(e)}')
 
@@ -87,12 +90,12 @@ def create_mapping_templates(client):
                 mapping = json.load(f)
 
                 template_name = os.path.splitext(filename)[0]  # Remove the .mapping extension
-                print(f'About to created index template: {template_name}')
+                logger.info(f'About to created index template: {template_name}')
 
                 # Create the template
                 try:
                     client.indices.put_index_template(name=template_name, body=mapping)
-                    print(f'Successfully created index template: {template_name}')
+                    logger.info(f'Successfully created index template: {template_name}')
                 except OpenSearchException as e:
                     logger.error(f'Failed to create index template: {template_name}, error: {str(e)}')
 
@@ -143,10 +146,28 @@ def load_dashboards():
 # create the data_streams based on the list given in the data-stream.json file
 def create_datasources():
     datasource_dir = '../datasource/'
+    # get current list of data sources
+    try:
+        response = requests.get(
+            url=f'https://{opensearch_host}:9200/_plugins/_query/_datasources',
+            auth=HTTPBasicAuth('admin', 'admin'),
+            verify=False,  # Disable SSL verification
+            headers={'Content-Type': 'application/json'}
+        )
+        response.raise_for_status()  # Raise an exception if the request failed
+        current_datasources = response.json()  # convert response to json
+    except requests.HTTPError as e:
+        logger.error(f'Failed to fetch datasources, error: {str(e)}')
+
     for filename in os.listdir(datasource_dir):
         if filename.endswith('.json'):
             with open(os.path.join(datasource_dir, filename), 'r') as f:
                 datasource = json.load(f)
+                # check if datasource already exists
+                if any(d['name'] == datasource['name'] for d in current_datasources):
+                    logger.info(f'Datasource already exists: {filename}')
+                    continue  # Skip to the next datasource if this one already exists
+
                 try:
                     response = requests.post(
                         url=f'https://{opensearch_host}:9200/_plugins/_query/_datasources',
@@ -157,12 +178,11 @@ def create_datasources():
                     )
                     response.raise_for_status()  # Raise an exception if the request failed
                     logger.info(f'Successfully created datasource: {filename}')
-                    print(f'Successfully created datasource: {filename}')
                 except requests.HTTPError as e:
-                    print(f'Failed to create datasource: {filename}, error: {str(e)}')
                     logger.error(f'Failed to create datasource: {filename}, error: {str(e)}')
 
 
+
 if __name__ == '__main__':
     # import all assets
     test_connection()

src/otelcollector/README.md

@@ -0,0 +1,129 @@
+# OTEL Collector Pipeline
+
+The next diagram describes the Observability signals pipeline defined inside the OTEL collector
+
+```mermaid
+%%{
+  init: {
+    'theme': 'base',
+    'themeVariables': {
+      'primaryColor': '#BB2528',
+      'primaryTextColor': '#fff',
+      'primaryBorderColor': '#7C0000',
+      'lineColor': '#F8B229',
+      'secondaryColor': '#006100',
+      'tertiaryColor': '#fff'
+    }
+  }
+}%%
+
+graph LR
+    otlp-->otlpReceiverTraces["otlp Receiver"]
+    style otlpReceiverTraces fill:#f9d,stroke:#333,stroke-width:4px
+    otlp-->otlpReceiverMetrics["otlp Receiver"]
+    style otlpReceiverMetrics fill:#f9d,stroke:#333,stroke-width:4px
+    otlp-->otlpReceiverLogs["otlp Receiver"]
+    style otlpReceiverLogs fill:#f9d,stroke:#333,stroke-width:4px
+    otlpServiceGraph-->otlpServiceGraphReceiver["otlp/servicegraph Receiver"]
+    spanmetrics-->spanmetricsReceiver["spanmetrics Receiver"]
+
+    subgraph Traces Pipeline
+    otlpReceiverTraces-->tracesProcessor["Traces Processor"]
+    style tracesProcessor fill:#9cf,stroke:#333,stroke-width:4px
+    tracesProcessor-->otlpExporter["otlp Exporter"]
+    tracesProcessor-->loggingExporterTraces["Logging Exporter"]
+
+    tracesProcessor-->spanmetricsExporter["Spanmetrics Exporter"]
+    tracesProcessor-->otlp2Exporter["otlp/2 Exporter"]
+    end
+
+    subgraph Metrics/Servicegraph Pipeline
+    otlpServiceGraphReceiver-->metricsServiceGraphProcessor["Metrics/ServiceGraph Processor"]
+    style otlpServiceGraphReceiver fill:#f9d,stroke:#333,stroke-width:4px
+    metricsServiceGraphProcessor-->prometheusServiceGraphExporter["Prometheus/ServiceGraph Exporter"]
+    style metricsServiceGraphProcessor fill:#9cf,stroke:#333,stroke-width:4px
+    end
+
+    subgraph Metrics Pipeline
+    otlpReceiverMetrics-->metricsProcessor["Metrics Processor"]
+    style metricsProcessor fill:#9cf,stroke:#333,stroke-width:4px
+
+    spanmetricsReceiver-->metricsProcessor
+    style spanmetricsReceiver fill:#f9d,stroke:#333,stroke-width:4px
+    metricsProcessor-->prometheusExporter["Prometheus Exporter"]
+    metricsProcessor-->loggingExporterMetrics["Logging Exporter"]
+    end
+
+    subgraph Logs Pipeline
+    otlpReceiverLogs-->logsProcessor["Logs Processor"]
+    style logsProcessor fill:#9cf,stroke:#333,stroke-width:4px
+    logsProcessor-->loggingExporterLogs["Logging Exporter"]
+    end
+
+
+```
+
+### Traces
+The traces  pipeline consists of a receiver, multiple processors, and multiple exporters.
+
+**Receiver (otlp):**
+This is where the data comes in from. In your configuration, the traces pipeline is using the otlp receiver. OTLP stands for OpenTelemetry Protocol. This receiver is configured to accept data over both gRPC and HTTP protocols. The HTTP protocol is also configured to allow CORS from any origin.
+
+**Processors (memory_limiter, batch, servicegraph):**
+Once the data is received, it is processed before being exported. The processors in the traces pipeline are:
+
+1. **memory_limiter:** This processor checks memory usage every second (check_interval: 1s) and ensures it does not exceed 4000 MiB (limit_mib: 4000). It also allows for a spike limit of 800 MiB (spike_limit_mib: 800).
+
+2. **batch:** This processor batches together traces before sending them on to the exporters, improving efficiency.
+
+3. **servicegraph:** This processor is specifically designed for creating a service graph from the traces. It is configured with certain parameters for handling latency histogram buckets, dimensions, store configurations, and so on.
+
+**Exporters (otlp, logging, spanmetrics, otlp/2):**
+After processing, the data is sent to the configured exporters:
+
+1. **otlp:** This exporter sends data to an endpoint configured as jaeger:4317 over OTLP with TLS encryption in insecure mode.
+
+2. **logging:** This exporter logs the traces.
+
+3. **spanmetrics:** This is likely a custom exporter defined as a connector in your configuration. It seems to be designed to create metrics from spans, but this is mostly conjecture based on the name.
+
+4. **otlp/2:** This exporter sends data to an endpoint configured as data-prepper:21890 over OTLP with TLS encryption in insecure mode.
+
+### Metrics
+**Metrics Pipeline**
+
+This pipeline handles metric data.
+
+- **Receivers (otlp, spanmetrics):**
+
+Metric data comes in from the `otlp` receiver and the `spanmetrics` receiver.
+- **Processors (filter, memory_limiter, batch):**
+The data is then processed:
+1. **filter:** This processor excludes specific metrics. In this configuration, it is set to strictly exclude the queueSize metric.
+2. **memory_limiter:** Similar to the traces pipeline, this processor ensures memory usage doesn't exceed a certain limit.
+3. **batch:** This processor batches together metrics before sending them to the exporters, enhancing efficiency.
+
+- **Exporters (prometheus, logging):**
+The processed data is then exported:
+1. **prometheus:** This exporter sends metrics to an endpoint configured as
+2. **otelcol:9464**. It also converts resource information to Prometheus labels and enables OpenMetrics.
+3. **logging:** This exporter logs the metrics.
+
+### Logs
+
+**Logs Pipeline**
+
+This pipeline handles log data.
+
+- **Receiver (otlp):**
+
+    Log data comes in from the otlp receiver. 
+- **Processors (memory_limiter, batch):**
+The data is then processed:
+1. **memory_limiter:** Similar to the traces and metrics pipelines, this processor ensures memory usage doesn't exceed a certain limit.
+2. **batch:** This processor batches together logs before sending them to the exporter, enhancing efficiency.
+
+- **Exporter (logging):**
+
+The processed data is then exported:
+ -  Logs Pipeline

src/otelcollector/otelcol-config.yml

@@ -11,6 +11,10 @@ receivers:
           allowed_origins:
             - "http://*"
             - "https://*"
+  otlp/servicegraph: # Dummy receiver for the metrics pipeline
+    protocols:
+      grpc:
+        endpoint: localhost:12345
 
 exporters:
   otlp/2:
@@ -28,30 +32,57 @@ exporters:
     resource_to_telemetry_conversion:
       enabled: true
     enable_open_metrics: true
+  prometheus/servicegraph:
+    endpoint: localhost:9090
+    namespace: servicegraph
 
 processors:
   batch:
+  memory_limiter:
+    check_interval: 1s
+    limit_mib: 4000
+    spike_limit_mib: 800
   filter:
     metrics:
       exclude:
         match_type: strict
         metric_names:
           - queueSize
+  servicegraph:
+    metrics_exporter: prometheus/servicegraph # Exporter to send metrics to
+    latency_histogram_buckets: [100us, 1ms, 2ms, 6ms, 10ms, 100ms, 250ms] # Buckets for latency histogram
+    dimensions: [cluster, namespace] # Additional dimensions (labels) to be added to the metrics extracted from the resource and span attributes
+    store: # Configuration for the in-memory store
+      ttl: 2s # Value to wait for an edge to be completed
+      max_items: 200 # Amount of edges that will be stored in the storeMap
+    cache_loop: 2m # the time to cleans the cache periodically
+    store_expiration_loop: 10s # the time to expire old entries from the store periodically.
+    virtual_node_peer_attributes:
+      - db.name
+      - rpc.service
 
 connectors:
   spanmetrics:
 
+extensions:
+  health_check: {}
+
 service:
+  extensions: [health_check]
   pipelines:
     traces:
       receivers: [otlp]
-      processors: [batch]
+      processors: [memory_limiter, batch, servicegraph]
       exporters: [otlp, logging, spanmetrics, otlp/2]
+    metrics/servicegraph:
+      receivers: [otlp/servicegraph]
+      processors: []
+      exporters: [prometheus/servicegraph]
     metrics:
       receivers: [otlp, spanmetrics]
-      processors: [filter, batch]
+      processors: [filter, memory_limiter, batch]
       exporters: [prometheus, logging]
     logs:
       receivers: [otlp]
-      processors: [batch]
+      processors: [memory_limiter, batch]
       exporters: [logging]