Add logs for Ad service and Recommendation service (open-telemetry#796)

* What’s included?

- Enable logs for Ad service and Recommendation service.
- Add OTLP exporters for logs
- Add the filter processor to prevent an error from the Prometheus exporter for duplicate queueSize metric, see
open-telemetry/opentelemetry-collector-contrib#18194. The filter processor can be removed when the fault gets fixed.

This PR doesn’t introduce any logs backend. Instead, logs are output only to Logging exporter and can be seen in the console (otelcol).
otel-col  | 2023-03-17T11:40:22.662Z	info	LogsExporter	{"kind": "exporter", "data_type": "logs", "name": "logging", "#logs": 2}

After this PR, different logging backends can be easily tested by configuring an additional exporter.

* Add changelog and fix lint errors.

* Fix changelog and lint

* Fix lint

* Move protocol env variables to .env

* Update CHANGELOG.md

---------

Co-authored-by: Austin Parker <austin@ap2.io>
Co-authored-by: Juliano Costa <julianocosta89@outlook.com>

.env

@@ -25,8 +25,14 @@ OTEL_COLLECTOR_PORT=4317
 OTEL_EXPORTER_OTLP_ENDPOINT=http://${OTEL_COLLECTOR_HOST}:${OTEL_COLLECTOR_PORT}
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT}
 OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT}
+OTEL_EXPORTER_OTLP_LOGS_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT}
 PUBLIC_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:4318/v1/traces
 
+# Exporter protocol configuration
+OTEL_TRACES_EXPORTER=otlp
+OTEL_METRICS_EXPORTER=otlp
+OTEL_LOGS_EXPORTER=otlp
+
 # OpenTelemetry Resource Definitions
 OTEL_RESOURCE_ATTRIBUTES="service.namespace=opentelemetry-demo"
 

CHANGELOG.md

@@ -11,8 +11,12 @@ release.
 ([#787](https://github.com/open-telemetry/opentelemetry-demo/pull/787))
 * [cart] use 60m TTL for cart entries in redis
 ([#779](https://github.com/open-telemetry/opentelemetry-demo/pull/779))
+* [bug] fixing quoteservice metrics exporting (PHP)
+([#793](https://github.com/open-telemetry/opentelemetry-demo/pull/793))
 * Added app.session.id attribute to frontend spans
 ([#795](https://github.com/open-telemetry/opentelemetry-demo/pull/795))
+* Add logs for Ad service and Recommendation service
+([#796](https://github.com/open-telemetry/opentelemetry-demo/pull/796))
 
 ## v0.1.0
 
@@ -275,5 +279,3 @@ significant modifications will be credited to OpenTelemetry Authors.
   ([#764](https://github.com/open-telemetry/opentelemetry-demo/pull/764))
 * [chore] align memory limits with Helm chart
   ([#781](https://github.com/open-telemetry/opentelemetry-demo/pull/781))
-* [bug] fixing quoteservice metrics exporting (PHP)
-  ([#793](https://github.com/open-telemetry/opentelemetry-demo/pull/793))

docker-compose.yml

@@ -78,8 +78,10 @@ services:
       - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
       - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
       - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
+      - OTEL_EXPORTER_OTLP_LOGS_ENDPOINT
       - OTEL_RESOURCE_ATTRIBUTES
       - OTEL_SERVICE_NAME=adservice
+      - OTEL_LOGS_EXPORTER
     depends_on:
       - otelcol
     logging: *logging
@@ -485,8 +487,8 @@ services:
       - PRODUCT_CATALOG_SERVICE_ADDR
       - FEATURE_FLAG_GRPC_SERVICE_ADDR
       - OTEL_PYTHON_LOG_CORRELATION=true
-      - OTEL_TRACES_EXPORTER=otlp
-      - OTEL_METRICS_EXPORTER=otlp
+      - OTEL_TRACES_EXPORTER
+      - OTEL_METRICS_EXPORTER
       - OTEL_EXPORTER_OTLP_ENDPOINT
       - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
       - OTEL_RESOURCE_ATTRIBUTES

src/adservice/Dockerfile

@@ -30,7 +30,7 @@ RUN ./gradlew installDist -PprotoSourceDir=./proto
 
 FROM eclipse-temurin:17-jre
 
-ARG version=1.23.0
+ARG version=1.24.0
 WORKDIR /usr/src/app/
 
 COPY --from=builder /usr/src/app/ ./

src/adservice/src/main/java/oteldemo/AdService.java

@@ -99,7 +99,7 @@ private void start() throws IOException {
             .addService(healthMgr.getHealthService())
             .build()
             .start();
-    logger.info("Ad Service started, listening on " + port);
+    logger.info("Ad service started, listening on " + port);
     Runtime.getRuntime()
         .addShutdownHook(
             new Thread(
@@ -160,15 +160,16 @@ public void getAds(AdRequest req, StreamObserver<AdResponse> responseObserver) {
 
         span.setAttribute("app.ads.contextKeys", req.getContextKeysList().toString());
         span.setAttribute("app.ads.contextKeys.count", req.getContextKeysCount());
-        logger.info("received ad request (context_words=" + req.getContextKeysList() + ")");
         if (req.getContextKeysCount() > 0) {
+          logger.info("Targeted ad request received for " + req.getContextKeysList());
           for (int i = 0; i < req.getContextKeysCount(); i++) {
             Collection<Ad> ads = service.getAdsByCategory(req.getContextKeys(i));
             allAds.addAll(ads);
           }
           adRequestType = AdRequestType.TARGETED;
           adResponseType = AdResponseType.TARGETED;
         } else {
+          logger.info("Non-targeted ad request received, preparing random response.");
           allAds = service.getRandomAds();
           adRequestType = AdRequestType.NOT_TARGETED;
           adResponseType = AdResponseType.RANDOM;
@@ -314,7 +315,7 @@ private static ImmutableListMultimap<String, Ad> createAdsMap() {
   /** Main launches the server from the command line. */
   public static void main(String[] args) throws IOException, InterruptedException {
     // Start the RPC server. You shouldn't see any output from gRPC before this.
-    logger.info("AdService starting.");
+    logger.info("Ad service starting.");
     final AdService service = AdService.getInstance();
     service.start();
     service.blockUntilShutdown();

src/otelcollector/otelcol-config.yml

@@ -43,14 +43,24 @@ processors:
       - context: metric
         statements:
           - set(description, "Measures the duration of inbound HTTP requests") where name == "http.server.duration"
+  filter:
+    metrics:
+      exclude:
+        match_type: strict
+        metric_names:
+          - queueSize
 
 service:
   pipelines:
     traces:
       receivers: [otlp]
       processors: [spanmetrics, batch]
-      exporters: [logging, otlp]
+      exporters: [otlp, logging]
     metrics:
       receivers: [otlp]
-      processors: [transform, batch]
+      processors: [filter, transform, batch]
       exporters: [prometheus, logging]
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [logging]

src/recommendationservice/recommendation_server.py

@@ -22,14 +22,20 @@
 # Pip
 import grpc
 from opentelemetry import trace, metrics
-
+from opentelemetry._logs import set_logger_provider
+from opentelemetry.exporter.otlp.proto.grpc._log_exporter import (
+    OTLPLogExporter,
+)
+from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.sdk.resources import Resource
 
 # Local
+import logging
 import demo_pb2
 import demo_pb2_grpc
 from grpc_health.v1 import health_pb2
 from grpc_health.v1 import health_pb2_grpc
-from logger import getJSONLogger
 
 from metrics import (
     init_metrics
@@ -43,7 +49,8 @@ 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(f"[Recv ListRecommendations] product_ids={prod_list}")
+        logger.info(f"Receive ListRecommendations for product ids:{prod_list}")
+
         # build and return response
         response = demo_pb2.ListRecommendationsResponse()
         response.product_ids.extend(prod_list)
@@ -78,15 +85,15 @@ def get_product_list(request_product_ids):
             if random.random() < 0.5 or first_run:
                 first_run = False
                 span.set_attribute("app.cache_hit", False)
-                logger.info("cache miss")
+                logger.info("get_product_list: cache miss")
                 cat_response = product_catalog_stub.ListProducts(demo_pb2.Empty())
                 response_ids = [x.id for x in cat_response.products]
                 cached_ids = cached_ids + response_ids
                 cached_ids = cached_ids + cached_ids[:len(cached_ids) // 4]
                 product_ids = cached_ids
             else:
                 span.set_attribute("app.cache_hit", True)
-                logger.info("cache hit")
+                logger.info("get_product_list: cache hit")
                 product_ids = cached_ids
         else:
             span.set_attribute("app.recommendation.cache_enabled", False)
@@ -110,7 +117,6 @@ def get_product_list(request_product_ids):
 
         return prod_list
 
-
 def must_map_env(key: str):
     value = os.environ.get(key)
     if value is None:
@@ -119,19 +125,35 @@ def must_map_env(key: str):
 
 def check_feature_flag(flag_name: str):
     flag = feature_flag_stub.GetFlag(demo_pb2.GetFlagRequest(name=flag_name)).flag
-    logger.info(flag)
     return flag.enabled
 
 if __name__ == "__main__":
+    service_name = must_map_env('OTEL_SERVICE_NAME')
+
     # Initialize Traces and Metrics
-    tracer = trace.get_tracer_provider().get_tracer("recommendationservice")
-    meter = metrics.get_meter_provider().get_meter("recommendationservice")
+    tracer = trace.get_tracer_provider().get_tracer(service_name)
+    meter = metrics.get_meter_provider().get_meter(service_name)
     rec_svc_metrics = init_metrics(meter)
 
-    port = must_map_env('RECOMMENDATION_SERVICE_PORT')
+    # Initialize Logs
+    logger_provider = LoggerProvider(
+        resource=Resource.create(
+            {
+                'service.name': service_name,
+            }
+        ),
+    )
+    set_logger_provider(logger_provider)
+    log_exporter = OTLPLogExporter(insecure=True)
+    logger_provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter))
+    handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
+
+    # Attach OTLP handler to logger
+    logger = logging.getLogger('main')
+    logger.addHandler(handler)
+
     catalog_addr = must_map_env('PRODUCT_CATALOG_SERVICE_ADDR')
     ff_addr = must_map_env('FEATURE_FLAG_GRPC_SERVICE_ADDR')
-
     pc_channel = grpc.insecure_channel(catalog_addr)
     ff_channel = grpc.insecure_channel(ff_addr)
     product_catalog_stub = demo_pb2_grpc.ProductCatalogServiceStub(pc_channel)
@@ -145,11 +167,9 @@ def check_feature_flag(flag_name: str):
     demo_pb2_grpc.add_RecommendationServiceServicer_to_server(service, server)
     health_pb2_grpc.add_HealthServicer_to_server(service, server)
 
-    # Start logger
-    logger = getJSONLogger('recommendationservice-server')
-    logger.info(f"RecommendationService listening on port: {port}")
-
     # Start server
+    port = must_map_env('RECOMMENDATION_SERVICE_PORT')
     server.add_insecure_port(f'[::]:{port}')
     server.start()
+    logger.info(f'Recommendation service started, listening on port {port}')
     server.wait_for_termination()