Implement message queue

.env

@@ -80,4 +80,9 @@ GRAFANA_SERVICE_HOST=grafana
 JAEGER_SERVICE_PORT=16686
 JAEGER_SERVICE_HOST=jaeger
 
+# Kafka
+KAFKA_SERVICE_PORT=9092
+KAFKA_SERVICE_ADDR=broker:${KAFKA_SERVICE_PORT}
+ZOOKEEPER_SERVICE_PORT=2181
+
 ENV_PLATFORM=local

CHANGELOG.md

@@ -146,3 +146,5 @@ significant modifications will be credited to OpenTelemetry Authors.
 ([#600](https:/open-telemetry/opentelemetry-demo/pull/600))
 * Add HTTP client instrumentation to shippingservice
 ([#610](https:/open-telemetry/opentelemetry-demo/pull/610))
+* Added Kafka, accountingservice and frauddetectionservice for async workflows
+ ([#512](https:/open-telemetry/opentelemetry-demo/pull/457))

docker-compose.yml

@@ -25,7 +25,32 @@ networks:
  driver: bridge
 
 services:
- # Ad service
+ # Accounting service
+ accountingservice:
+ image: ${IMAGE_NAME}:${IMAGE_VERSION}-accountingservice
+ container_name: accounting-service
+ build:
+ context: ./
+ dockerfile: ./src/accountingservice/Dockerfile
+ cache_from:
+ - ${IMAGE_NAME}:${IMAGE_VERSION}-accountingservice
+ deploy:
+ resources:
+ limits:
+ memory: 300M
+ restart: always
+ environment:
+ - KAFKA_SERVICE_ADDR
+ - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
+ - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
+ - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
+ - OTEL_SERVICE_NAME=accountingservice
+ depends_on:
+ - otelcol
+ - broker
+ logging: *logging
+
+ # AdService
  adservice:
  image: ${IMAGE_NAME}:${IMAGE_VERSION}-adservice
  container_name: ad-service
@@ -102,6 +127,7 @@ services:
  - PAYMENT_SERVICE_ADDR
  - PRODUCT_CATALOG_SERVICE_ADDR
  - SHIPPING_SERVICE_ADDR
+ - KAFKA_SERVICE_ADDR
  - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
  - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
  - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
@@ -114,6 +140,7 @@ services:
  - productcatalogservice
  - shippingservice
  - otelcol
+ - broker
  logging: *logging
 
  # Currency service
@@ -166,6 +193,30 @@ services:
  - otelcol
  logging: *logging
 
+ frauddetection:
+ image: ${IMAGE_NAME}:${IMAGE_VERSION}-frauddetectionservice
+ container_name: frauddetection-service
+ build:
+ context: ./
+ dockerfile: ./src/frauddetectionservice/Dockerfile
+ cache_from:
+ - ${IMAGE_NAME}:${IMAGE_VERSION}-frauddetectionservice
+ deploy:
+ resources:
+ limits:
+ memory: 200M
+ restart: always
+ environment:
+ - KAFKA_SERVICE_ADDR
+ - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
+ - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
+ - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
+ - OTEL_SERVICE_NAME=frauddetectionservice
+ depends_on:
+ - otelcol
+ - broker
+ logging: *logging
+
  # Feature Flag service
  featureflagservice:
  image: ${IMAGE_NAME}:${IMAGE_VERSION}-featureflagservice
@@ -444,7 +495,7 @@ services:
  deploy:
  resources:
  limits:
- memory: 120M
+ memory: 200M
  restart: always
  environment:
  - POSTGRES_USER=ffs
@@ -538,6 +589,29 @@ services:
  - "${REDIS_PORT}"
  logging: *logging
 
+ broker:
+ image: confluentinc/cp-kafka:7.2.2-1-ubi8
+ hostname: broker
+ container_name: broker
+ environment:
+ KAFKA_LISTENERS: PLAINTEXT://broker:9092,CONTROLLER://broker:9093
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092
+ KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+ KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+ KAFKA_PROCESS_ROLES: 'controller,broker'
+ KAFKA_NODE_ID: 1
+ KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:9093'
+ KAFKA_METADATA_LOG_SEGMENT_MS: 15000
+ KAFKA_METADATA_MAX_RETENTION_MS: 1200000
+ KAFKA_METADATA_LOG_MAX_RECORD_BYTES_BETWEEN_SNAPSHOTS: 2800
+ KAFKA_AUTO_CREATE_TOPICS_ENABLE: true
+ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+ volumes:
+ - ./src/kafka/update_run.sh:/tmp/update_run.sh
+ - ./src/kafka/clusterID:/tmp/clusterID
+ command: "bash -c 'ls -lh /tmp && /tmp/update_run.sh && /etc/confluent/docker/run'"
 
  # Frontend Tests
  frontendTests:

docs/README.md

@@ -24,18 +24,19 @@ Want to deploy the demo and see it in action? Start here.
 Want to understand how a particular language's instrumentation works? Start
 here.
 
-| Language | Auto Instrumentation | Manual Instrumentation |
-|---------------|---------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------|
-| .NET | [Cart Service](./services/cartservice.md) | [Cart Service](./services/cartservice.md) |
-| C++ | | [Currency Service](./services/currencyservice.md) |
-| Erlang/Elixir | [Feature Flag Service](./services/featureflagservice.md) | [Feature Flag Service](./services/featureflagservice.md) |
-| Go | [Checkout Service](./services/checkoutservice.md), [Product Catalog Service]( ./services/productcatalogservice.md ) | [Checkout Service](./services/checkoutservice.md), [Product Catalog Service]( ./services/productcatalogservice.md ) |
-| Java | [Ad Service](./services/adservice.md) | [Ad Service](./services/adservice.md) |
-| JavaScript | [Frontend]( ./services/frontend.md ) | [Frontend](./services/frontend.md), [Payment Service](./services/paymentservice.md) |
-| PHP | [Quote Service](./services/quoteservice.md) | [Quote Service](./services/quoteservice.md) |
-| Python | [Recommendation Service](./services/recommendationservice.md) | [Recommendation Service](./services/recommendationservice.md) |
-| Ruby | [Email Service](./services/emailservice.md) | [Email Service](./services/emailservice.md) |
-| Rust | [Shipping Service](./services/shippingservice.md) | [Shipping Service](./services/shippingservice.md) |
+| Language | Auto Instrumentation | Manual Instrumentation |
+|---------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------|
+| .NET | [Cart Service](./services/cartservice.md) | [Cart Service](./services/cartservice.md) |
+| C++ | | [Currency Service](./services/currencyservice.md) |
+| Erlang/Elixir | [Feature Flag Service](./services/featureflagservice.md) | [Feature Flag Service](./services/featureflagservice.md) |
+| Go | [Accounting Service](./services/accountingservice.md), [Checkout Service](./services/checkoutservice.md), [Product Catalog Service]( ./services/productcatalogservice.md ) | [Checkout Service](./services/checkoutservice.md), [Product Catalog Service]( ./services/productcatalogservice.md ) |
+| Java | [Ad Service](./services/adservice.md) | [Ad Service](./services/adservice.md) |
+| JavaScript | [Frontend]( ./services/frontend.md ) | [Frontend](./services/frontend.md), [Payment Service](./services/paymentservice.md) |
+| Kotlin | [Fraud Detection Service]( ./services/frauddetectionservice.md ) | |
+| PHP | [Quote Service](./services/quoteservice.md) | [Quote Service](./services/quoteservice.md) |
+| Python | [Recommendation Service](./services/recommendationservice.md) | [Recommendation Service](./services/recommendationservice.md) |
+| Ruby | [Email Service](./services/emailservice.md) | [Email Service](./services/emailservice.md) |
+| Rust | [Shipping Service](./services/shippingservice.md) | [Shipping Service](./services/shippingservice.md) |
 
 ### Service Documentation
 

docs/current_architecture.md

@@ -8,11 +8,13 @@ uses [Locust](https://locust.io/) to fake user traffic.
 graph TD
 subgraph Service Diagram
 adservice(Ad Service):::java
+accountingservice(Accounting Service):::golang
 cache[(Cache<br/>&#40redis&#41)]
 cartservice(Cart Service):::dotnet
 checkoutservice(Checkout Service):::golang
 currencyservice(Currency Service):::cpp
 emailservice(Email Service):::ruby
+frauddetectionservice(Fraud Detection Service):::kotlin
 frontend(Frontend):::javascript
 frontendproxy(Frontend Proxy <br/>&#40Envoy&#41):::cpp
 loadgenerator([Load Generator]):::python
@@ -23,18 +25,22 @@ recommendationservice(Recommendation Service):::python
 shippingservice(Shipping Service):::rust
 featureflagservice(Feature Flag Service):::erlang
 featureflagstore[(Feature Flag Store<br/>&#40PostgreSQL DB&#41)]
+queue[(queue<br/>&#40Kafka&#41)]
 
 Internet -->|HTTP| frontendproxy
 frontendproxy -->|HTTP| frontend
 frontendproxy -->|HTTP| featureflagservice
 loadgenerator -->|HTTP| frontend
 
+accountingservice -->|TCP| queue
+
 checkoutservice --->|gRPC| cartservice --> cache
 checkoutservice --->|gRPC| productcatalogservice
 checkoutservice --->|gRPC| currencyservice
 checkoutservice --->|HTTP| emailservice
 checkoutservice --->|gRPC| paymentservice
 checkoutservice -->|gRPC| shippingservice
+checkoutservice -->|TCP| queue
 
 frontend -->|gRPC| adservice
 frontend -->|gRPC| cartservice
@@ -44,6 +50,8 @@ frontend -->|gRPC| currencyservice
 frontend -->|gRPC| recommendationservice -->|gRPC| productcatalogservice
 frontend -->|gRPC| shippingservice -->|HTTP| quoteservice
 
+frauddetectionservice -->|TCP| queue
+
 productcatalogservice -->|gRPC| featureflagservice
 
 shippingservice -->|gRPC| featureflagservice
@@ -53,6 +61,7 @@ featureflagservice --> featureflagstore
 end
 
 classDef java fill:#b07219,color:white;
+classDef kotlin fill:#560ba1,color:white;
 classDef dotnet fill:#178600,color:white;
 classDef golang fill:#00add8,color:black;
 classDef cpp fill:#f34b7d,color:white;
@@ -77,9 +86,11 @@ subgraph Service Legend
  rustsvc(Rust):::rust
  erlangsvc(Erlang/Elixir):::erlang
  phpsvc(PHP):::php
+ kotlinsvc(Kotlin):::kotlin
 end
 
 classDef java fill:#b07219,color:white;
+classDef kotlin fill:#560ba1,color:white;
 classDef dotnet fill:#178600,color:white;
 classDef golang fill:#00add8,color:black;
 classDef cpp fill:#f34b7d,color:white;

docs/docker_deployment.md

@@ -4,7 +4,7 @@
 
 - Docker
 - [Docker Compose](https://docs.docker.com/compose/install/#install-compose) v2.0.0+
-- 4 GB of RAM
+- 5 GB of RAM
 
 ## Clone Repo
 

docs/service_table.md

@@ -4,11 +4,13 @@ View [Service Graph](./current_architecture.md) to visualize request flows.
 
 | Service | Language | Description |
 |--------------------------------------------------------------|---------------|----------------------------------------------------------------------------------------------------------------------------------------------|
+| [accountingservice](./services/accountingservice.md) | Go | Processes incoming orders and count the sum of all orders (mock). |
 | [adservice](./services/adservice.md) | Java | Provides text ads based on given context words. |
 | [cartservice](./services/cartservice.md) | DotNet | Stores the items in the user's shopping cart in Redis and retrieves it. |
 | [checkoutservice](./services/checkoutservice.md) | Go | Retrieves user cart, prepares order and orchestrates the payment, shipping and the email notification. |
 | [currencyservice](./services/currencyservice.md) | C++ | Converts one money amount to another currency. Uses real values fetched from European Central Bank. It's the highest QPS service. |
 | [emailservice](./services/emailservice.md) | Ruby | Sends users an order confirmation email (mock). |
+| [frauddetectionservice](./services/frauddetectionservice.md) | Kotlin | Analyzes incoming orders and detects fraud attempts (mock). |
 | [featureflagservice](./services/featureflagservice.md) | Erlang/Elixir | CRUD feature flag service to demonstrate various scenarios like fault injection & how to emit telemetry from a feature flag reliant service. |
 | [frontend](./services/frontend.md) | JavaScript | Exposes an HTTP server to serve the website. Does not require signup/login and generates session IDs for all users automatically. |
 | [loadgenerator](./services/loadgenerator.md) | Python/Locust | Continuously sends requests imitating realistic user shopping flows to the frontend. |

docs/services/accountingservice.md

@@ -0,0 +1,58 @@
+# Accounting Service
+
+This service calculates the total amount of sold products.
+This is only mocked and received orders are printed out.
+
+[Accounting Service](../../src/accountingservice/)
+
+## Traces
+
+### Initializing Tracing
+
+The OpenTelemetry SDK is initialized from `main` using the `initTracerProvider`
+function.
+
+```go
+func initTracerProvider() (*sdktrace.TracerProvider, error) {
+ ctx := context.Background()
+
+ exporter, err := otlptracegrpc.New(ctx)
+ if err != nil {
+ return nil, err
+ }
+ tp := sdktrace.NewTracerProvider(
+ sdktrace.WithBatcher(exporter),
+ )
+ otel.SetTracerProvider(tp)
+ otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
+ return tp, nil
+}
+```
+
+You should call `TracerProvider.Shutdown()` when your service is shutdown to
+ensure all spans are exported. This service makes that call as part of a
+deferred function in main
+
+```go
+ tp, err := initTracerProvider()
+ if err != nil {
+ log.Fatal(err)
+ }
+ defer func() {
+ if err := tp.Shutdown(context.Background()); err != nil {
+ log.Printf("Error shutting down tracer provider: %v", err)
+ }
+ }()
+```
+
+### Adding Kafka ( Sarama ) auto-instrumentation
+
+This service will receive the processed results of the Checkout Service via a
+Kafka topic.
+To instrument the Kafka client the ConsumerHandler implemented by the developer
+has to be wrapped.
+
+```go
+ handler := groupHandler{} // implements sarama.ConsumerGroupHandler
+ wrappedHandler := otelsarama.WrapConsumerGroupHandler(&handler)
+```

docs/services/checkoutservice.md

@@ -67,6 +67,21 @@ func createClient(ctx context.Context, svcAddr string) (*grpc.ClientConn, error)
 }
 ```
 
+### Adding Kafka ( Sarama ) auto-instrumentation
+
+This service will write the processed results onto a Kafka topic which will then
+be in turn be processed by other microservices.
+To instrument the Kafka client the Producer has to be wrapped after it has been created.
+
+```go
+ saramaConfig := sarama.NewConfig()
+ producer, err := sarama.NewAsyncProducer(brokers, saramaConfig)
+ if err != nil {
+ return nil, err
+ }
+ producer = otelsarama.WrapAsyncProducer(saramaConfig, producer)
+```
+
 ### Add attributes to auto-instrumented spans
 
 Within the execution of auto-instrumented code you can get current span from

docs/services/frauddetectionservice.md

@@ -0,0 +1,16 @@
+# Fraud Detection Service
+
+This service analyses incoming orders and detects malicious customers.
+This is only mocked and received orders are printed out.
+
+## Auto-instrumentation
+
+This service relies on the OpenTelemetry Java Agent to automatically instrument
+libraries such as Kafka, and to configure the OpenTelemetry SDK. The agent is
+passed into the process using the `-javaagent` command line argument. Command
+line arguments are added through the `JAVA_TOOL_OPTIONS` in the `Dockerfile`,
+and leveraged during the automatically generated Gradle startup script.
+
+```dockerfile
+ENV JAVA_TOOL_OPTIONS=-javaagent:/app/opentelemetry-javaagent.jar
+```