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

@@ -130,3 +130,5 @@ significant modifications will be credited to OpenTelemetry Authors.
 ([#513](https:/open-telemetry/opentelemetry-demo/pull/513))
 * Added frontend instrumentation exporter custom url
 ([#512](https:/open-telemetry/opentelemetry-demo/pull/512))
+* Added Kafka, accountingservice and frauddetectionservice for async workflows
+([#512](https:/open-telemetry/opentelemetry-demo/pull/457))

docker-compose.yml

@@ -26,6 +26,31 @@ networks:
 
 services:
  # Ad 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
+
  # Frontend
  frontend:
  image: ${IMAGE_NAME}:${IMAGE_VERSION}-frontend
@@ -534,6 +585,34 @@ services:
  - "${REDIS_PORT}"
  logging: *logging
 
+ # Kafka
+ zookeeper:
+ image: confluentinc/cp-zookeeper:7.2.2
+ container_name: zookeeper
+ environment:
+ ZOOKEEPER_CLIENT_PORT: ${ZOOKEEPER_SERVICE_PORT}
+ ZOOKEEPER_TICK_TIME: 2000
+
+ broker:
+ image: confluentinc/cp-kafka:7.2.2
+ container_name: broker
+ depends_on:
+ - zookeeper
+ ports:
+ # To learn about configuring Kafka for access across networks see
+ # https://www.confluent.io/blog/kafka-client-cannot-connect-to-broker-on-aws-on-docker-etc/
+ - "9092:9092"
+ environment:
+ KAFKA_BROKER_ID: 1
+ KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:${ZOOKEEPER_SERVICE_PORT}'
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://:${KAFKA_SERVICE_PORT},PLAINTEXT_INTERNAL://:29092
+ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+ KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+ KAFKA_AUTO_CREATE_TOPICS_ENABLE: true
+
+ KAFKA_LOG4J_ROOT_LOGLEVEL: DEBUG
 
  # 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++ | | |
-| 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++ | | |
+| 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/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 services 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
+```

src/accountingservice/Dockerfile

@@ -0,0 +1,38 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM golang:1.19.2-alpine AS builder
+RUN apk add build-base protoc
+WORKDIR /usr/src/app/
+
+# Restore dependencies
+COPY ./src/accountingservice/ ./
+COPY ./pb/ ./proto/
+RUN go mod download
+RUN go install google.golang.org/protobuf/cmd/[email protected]
+RUN go install google.golang.org/grpc/cmd/[email protected]
+
+# Build executable
+RUN protoc -I ./proto/ ./proto/demo.proto --go_out=./ --go-grpc_out=./
+RUN go build -o /go/bin/accountingservice/ ./
+
+# -----------------------------------------------------------------------------
+
+FROM alpine
+
+WORKDIR /usr/src/app/
+
+COPY --from=builder /go/bin/accountingservice/ ./
+
+ENTRYPOINT [ "./accountingservice" ]

src/accountingservice/README.md

@@ -0,0 +1,20 @@
+# Accounting Service
+
+This service consumes new orders from a Kafka topic.
+
+## Local Build
+
+To build the protos and the service binary, run:
+
+```sh
+protoc -I ../../pb/ ../../pb/demo.proto --go_out=./ --go-grpc_out=./
+go build -o /go/bin/checkoutservice/ ./
+```
+
+## Docker Build
+
+From the root directory, run:
+
+```sh
+docker compose build checkoutservice
+```