Integrate with existing metrics code

CHANGELOG.md

@@ -113,3 +113,4 @@ significant modifications will be credited to OpenTelemetry Authors.
 ([#339](https://github.com/open-telemetry/opentelemetry-demo/pull/339))
 * Added basic metrics support for recommendation service (Python)
 ([#416](https://github.com/open-telemetry/opentelemetry-demo/pull/416))
+* Update metrics for recommendation service (Python) [#432](https://github.com/open-telemetry/opentelemetry-demo/pull/432)

docker-compose.yml

@@ -331,10 +331,9 @@ services:
       - RECOMMENDATION_SERVICE_PORT
       - PRODUCT_CATALOG_SERVICE_ADDR
       - OTEL_PYTHON_LOG_CORRELATION=true
-      - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
-      # OTEL_EXPORTER_OTLP_METRICS_ENDPOINT # Not working for Python OTLP exporter
+      - OTEL_METRICS_EXPORTER=console,otlp
       - OTEL_EXPORTER_OTLP_ENDPOINT
-      - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
+      - OTEL_EXPORTER_OTLP_INSECURE=true
       - OTEL_SERVICE_NAME=recommendationservice
       - PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python
     logging: *logging

src/recommendationservice/Dockerfile

@@ -22,8 +22,10 @@ COPY ./pb/ ./proto/
 RUN pip install --upgrade pip
 RUN pip install -r ./requirements.txt
 
+RUN opentelemetry-bootstrap -a install
+
 # add files into working directory
-RUN python -m pip install grpcio-tools
+RUN python -m pip install grpcio-tools==1.48.2
 RUN python -m grpc_tools.protoc -I=./proto/ --python_out=./ --grpc_python_out=./ ./proto/demo.proto
 
 EXPOSE ${RECOMMENDATION_SERVICE_PORT}

src/recommendationservice/metrics.py

@@ -0,0 +1,100 @@
+#!/usr/bin/python
+
+from gc import callbacks
+from typing import Iterable
+import psutil
+
+from opentelemetry.metrics import (
+    CallbackOptions,
+    Observation,
+)
+
+# RAM usage
+def ram_usage_callback(options: CallbackOptions) -> Iterable[Observation]:
+    observations = []    
+    ram_percent = psutil.virtual_memory().percent
+    print(f"ram_percent: {ram_percent}")
+    labels = {"dimension": "value"}
+    observations.append(Observation(ram_percent, labels))
+
+    return observations
+
+# CPU Time callback
+def cpu_time_callback(options: CallbackOptions) -> Iterable[Observation]:
+    observations = []
+    with open("/proc/stat") as procstat:
+        procstat.readline()  # skip the first line
+        for line in procstat:
+            if not line.startswith("cpu"): break
+            cpu, *states = line.split()
+            observations.append(Observation(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
+            observations.append(Observation(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
+            observations.append(Observation(int(states[2]) // 100, {"cpu": cpu, "state": "system"}))
+            # ... other states
+    return observations
+
+# CPU usage
+def cpu_usage_callback(options: CallbackOptions) -> Iterable[Observation]:
+    observations = []    
+    for (number, percent) in enumerate(psutil.cpu_percent(percpu=True)):
+        print(f"cpu_number: {number}, cpu_percent: {percent}")
+        labels = {"cpu_number": str(number)}
+        observations.append(Observation(percent, labels))
+
+    return observations
+
+def observable_counter_callback(options: CallbackOptions) -> Iterable[Observation]:
+    yield Observation(1, {})
+
+
+def observable_up_down_counter_callback(
+    options: CallbackOptions,
+) -> Iterable[Observation]:
+    yield Observation(-10, {})
+
+
+def observable_gauge_callback(options: CallbackOptions) -> Iterable[Observation]:
+    yield Observation(9, {})
+
+
+def init_metrics(meter):
+
+    # Requests counter
+    list_recommendations_request_counter = meter.create_counter(
+        name="app.recommendations.request.counter",
+        description="number of requests to RecommendationService.ListRecommendations",
+        unit="requests"
+    )
+
+    # Recommendations counter
+    app_recommendations_counter = meter.create_counter(
+        'app.recommendations.counter', unit='recommendations', description="Counts the total number of given recommendations"
+    )
+
+    # CPU usage
+    cpu_usage = meter.create_observable_counter(
+        "cpu_usage",
+        callbacks=[cpu_usage_callback],
+        unit="%",
+        description="CPU usage"
+    )
+
+    # RAM usage
+    ram_usage = meter.create_observable_up_down_counter(
+        "ram_usage",
+        callbacks=[ram_usage_callback],
+        unit="1",
+        description="RAM usage"        
+    )
+
+    attributes = {"application.name": "otel-demo"}
+
+    rec_svc_metrics = {
+        "list_recommendations_request_counter": list_recommendations_request_counter,
+        "attributes": attributes,
+        "app_recommendations_counter": app_recommendations_counter,
+        "cpu_usage": cpu_usage,
+        "ram_usage": ram_usage
+    }
+
+    return rec_svc_metrics

src/recommendationservice/recommendation_server.py

@@ -17,23 +17,12 @@
 # Python
 import os
 import random
-import time
 from concurrent import futures
 
 # Pip
 import grpc
+from opentelemetry import trace, metrics
 
-# Traces
-from opentelemetry import trace
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import (BatchSpanProcessor)
-from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
-
-# Metrics
-from opentelemetry import metrics
-from opentelemetry.sdk.metrics import MeterProvider
-from opentelemetry.sdk.metrics.export import (PeriodicExportingMetricReader)
-from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
 
 # Local
 import demo_pb2
@@ -42,18 +31,24 @@
 from grpc_health.v1 import health_pb2_grpc
 from logger import getJSONLogger
 
+from metrics import (
+    init_metrics
+)
 
 class RecommendationService(demo_pb2_grpc.RecommendationServiceServicer):
     def ListRecommendations(self, request, context):
         prod_list = get_product_list(request.product_ids)
         span = trace.get_current_span()
         span.set_attribute("app.products_recommended.count", len(prod_list))
-        logger.info("[Recv ListRecommendations] product_ids={}".format(prod_list))
-        app_recommendations_counter.add(len(prod_list), {'recommendation.type': 'catalog'})
-
+        logger.info(f"[Recv ListRecommendations] product_ids={prod_list}")
         # build and return response
         response = demo_pb2.ListRecommendationsResponse()
         response.product_ids.extend(prod_list)
+
+        # Collect metrics on # requests to this service
+        rec_svc_metrics["list_recommendations_request_counter"].add(1, rec_svc_metrics["attributes"])
+        rec_svc_metrics["app_recommendations_counter"].add(len(prod_list), {'recommendation.type': 'catalog'})
+
         return response
 
     def Check(self, request, context):
@@ -88,6 +83,9 @@ def get_product_list(request_product_ids):
         indices = random.sample(range(num_products), num_return)
         # Fetch product ids from indices
         prod_list = [filtered_products[i] for i in indices]
+
+        span.set_attribute("app.filtered_products.list", prod_list)
+
         return prod_list
 
 
@@ -98,22 +96,10 @@ def must_map_env(key: str):
     return value
 
 if __name__ == "__main__":
-    # Initialize tracer provider
-    tracer_provider = TracerProvider()
-    trace.set_tracer_provider(tracer_provider)
-    tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
-    tracer = trace.get_tracer("recommendationservice")
-
-    # Initialize meter provider
-    metric_reader = PeriodicExportingMetricReader(OTLPMetricExporter())
-    provider = MeterProvider(metric_readers=[metric_reader])
-    metrics.set_meter_provider(provider)
-    meter = metrics.get_meter(__name__)
-
-    # Create counters
-    app_recommendations_counter = meter.create_counter(
-        'app.recommendations.counter', unit='recommendations', description="Counts the total number of given recommendations"
-    )
+    # Initialize Traces and Metrics
+    tracer = trace.get_tracer_provider().get_tracer("recommendationservice")
+    meter = metrics.get_meter_provider().get_meter("recommendationservice")
+    rec_svc_metrics = init_metrics(meter)
 
     port = must_map_env('RECOMMENDATION_SERVICE_PORT')
     catalog_addr = must_map_env('PRODUCT_CATALOG_SERVICE_ADDR')
@@ -131,9 +117,9 @@ def must_map_env(key: str):
 
     # Start logger
     logger = getJSONLogger('recommendationservice-server')
-    logger.info("RecommendationService listening on port: " + port)
+    logger.info(f"RecommendationService listening on port: {port}")
 
     # Start server
-    server.add_insecure_port('[::]:' + port)
+    server.add_insecure_port(f'[::]:{port}')
     server.start()
     server.wait_for_termination()

src/recommendationservice/requirements.txt

@@ -1,13 +1,8 @@
 google-api-core==2.4.0
 grpcio-health-checking==1.43.0
 grpcio==1.43.0
-opentelemetry-api==1.13.0
-opentelemetry-exporter-otlp-proto-grpc==1.13.0
-opentelemetry-instrumentation==0.34b0
-opentelemetry-instrumentation-grpc==0.34b0
-opentelemetry-instrumentation-urllib3==0.34b0
-opentelemetry-sdk==1.13.0
-python-dotenv==0.20.0
-python-json-logger==2.0.2
-requests==2.27.1
-urllib3==1.26.8
+opentelemetry-distro==0.34b0
+opentelemetry-exporter-otlp==1.13.0
+python-dotenv==0.21.0
+python-json-logger==2.0.4
+psutil==5.9.2 # Importing this will also import opentelemetry-instrumentation-system-metrics when running opentelemetry-bootstrap