store order to postgresql from accounting service (open-telemetry#2175)

* store order to postgresql from accounting service

* update on comment

* fix trace not closed

* Revert "fix trace not closed"

This reverts commit e191c3e.

* disable entity framework instrumentation

* clean

* group spans

* fix for linter

---------

Co-authored-by: Pierre Tessier <pierre@pierretessier.com>

Author: cuichenli

.licenserc.json

@@ -38,6 +38,8 @@
     ],
     "ignore": [
         "node_modules/",
+        "src/accounting/bin/",
+        "src/accounting/obj/",
         "/src/accounting/src/protos/",
         "src/cart/src/obj/",
         "src/cart/tests/obj/",

docker-compose.yml

@@ -37,6 +37,8 @@ services:
       - OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE
       - OTEL_RESOURCE_ATTRIBUTES
       - OTEL_SERVICE_NAME=accounting
+      - DB_CONNECTION_STRING=Host=postgresql;Username=otelu;Password=otelp;Database=otel
+      - OTEL_DOTNET_AUTO_TRACES_ENTITYFRAMEWORKCORE_INSTRUMENTATION_ENABLED=false
     depends_on:
       otel-collector:
         condition: service_started
@@ -686,6 +688,18 @@ services:
       retries: 10
     logging: *logging
 
+  # Postgresql used by Accounting service
+  postgresql:
+    image: ${POSTGRES_IMAGE}
+    ports:
+      - "5432"
+    environment:
+      POSTGRES_USER: root
+      POSTGRES_PASSWORD: otel
+      POSTGRES_DB: otel
+    volumes:
+      - ${PWD}/src/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
+
   # Valkey used by Cart service
   valkey-cart:
     image: ${VALKEY_IMAGE}

src/accounting/.dockerignore

@@ -0,0 +1,3 @@
+bin
+obj
+build

src/accounting/.gitignore

@@ -0,0 +1 @@
+proto

src/accounting/Accounting.csproj

@@ -10,13 +10,15 @@
 
 	<ItemGroup>
 		<PackageReference Include="Confluent.Kafka" Version="2.8.0" />
+		<PackageReference Include="EFCore.NamingConventions" Version="9.0.0" />
 		<PackageReference Include="Google.Protobuf" Version="3.30.0" />
 		<PackageReference Include="Grpc.Tools" Version="2.68.1">
 			<PrivateAssets>all</PrivateAssets>
 			<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
 		</PackageReference>
 		<PackageReference Include="Microsoft.Extensions.Logging" Version="9.0.2" />
 		<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.21.2" />
+		<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="9.0.4" />
 		<PackageReference Include="OpenTelemetry.AutoInstrumentation" Version="1.11.0" />
 	</ItemGroup>
 

src/accounting/Consumer.cs

@@ -4,16 +4,35 @@
 using Confluent.Kafka;
 using Microsoft.Extensions.Logging;
 using Oteldemo;
+using Microsoft.EntityFrameworkCore;
+using System.Diagnostics;
 
 namespace Accounting;
 
+internal class DBContext : DbContext
+{
+    public DbSet<OrderEntity> Orders { get; set; }
+    public DbSet<OrderItemEntity> CartItems { get; set; }
+    public DbSet<ShippingEntity> Shipping { get; set; }
+
+    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
+    {
+        var connectionString = Environment.GetEnvironmentVariable("DB_CONNECTION_STRING");
+
+        optionsBuilder.UseNpgsql(connectionString).UseSnakeCaseNamingConvention();
+    }
+}
+
+
 internal class Consumer : IDisposable
 {
     private const string TopicName = "orders";
 
     private ILogger _logger;
     private IConsumer<string, byte[]> _consumer;
     private bool _isListening;
+    private DBContext? _dbContext;
+    private static readonly ActivitySource MyActivitySource = new("Accounting.Consumer");
 
     public Consumer(ILogger<Consumer> logger)
     {
@@ -26,6 +45,7 @@ public Consumer(ILogger<Consumer> logger)
         _consumer.Subscribe(TopicName);
 
         _logger.LogInformation($"Connecting to Kafka: {servers}");
+        _dbContext = Environment.GetEnvironmentVariable("DB_CONNECTION_STRING") == null ? null : new DBContext();
     }
 
     public void StartListening()
@@ -38,8 +58,8 @@ public void StartListening()
             {
                 try
                 {
+                    using var activity = MyActivitySource.StartActivity("order-consumed",  ActivityKind.Internal);
                     var consumeResult = _consumer.Consume();
-
                     ProcessMessage(consumeResult.Message);
                 }
                 catch (ConsumeException e)
@@ -61,8 +81,48 @@ private void ProcessMessage(Message<string, byte[]> message)
         try
         {
             var order = OrderResult.Parser.ParseFrom(message.Value);
-
             Log.OrderReceivedMessage(_logger, order);
+
+            if (_dbContext == null)
+            {
+                return;
+            }
+
+            var orderEntity = new OrderEntity
+            {
+                Id = order.OrderId
+            };
+            _dbContext.Add(orderEntity);
+            foreach (var item in order.Items)
+            {
+                var orderItem = new OrderItemEntity
+                {
+                    ItemCostCurrencyCode = item.Cost.CurrencyCode,
+                    ItemCostUnits = item.Cost.Units,
+                    ItemCostNanos = item.Cost.Nanos,
+                    ProductId = item.Item.ProductId,
+                    Quantity = item.Item.Quantity,
+                    OrderId = order.OrderId
+                };
+
+                _dbContext.Add(orderItem);
+            }
+
+            var shipping = new ShippingEntity
+            {
+                ShippingTrackingId = order.ShippingTrackingId,
+                ShippingCostCurrencyCode = order.ShippingCost.CurrencyCode,
+                ShippingCostUnits = order.ShippingCost.Units,
+                ShippingCostNanos = order.ShippingCost.Nanos,
+                StreetAddress = order.ShippingAddress.StreetAddress,
+                City = order.ShippingAddress.City,
+                State = order.ShippingAddress.State,
+                Country = order.ShippingAddress.Country,
+                ZipCode = order.ShippingAddress.ZipCode,
+                OrderId = order.OrderId
+            };
+            _dbContext.Add(shipping);
+            _dbContext.SaveChanges();
         }
         catch (Exception ex)
         {

src/accounting/Dockerfile

@@ -31,5 +31,6 @@ RUN mkdir -p "/var/log/opentelemetry/dotnet"
 RUN chown app "/var/log/opentelemetry/dotnet"
 RUN chown app "/app/instrument.sh"
 USER app
+ENV OTEL_DOTNET_AUTO_TRACES_ADDITIONAL_SOURCES=Accounting.Consumer
 
 ENTRYPOINT ["./instrument.sh", "dotnet", "Accounting.dll"]

src/accounting/Entities.cs

@@ -0,0 +1,59 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using Microsoft.EntityFrameworkCore;
+using System.ComponentModel.DataAnnotations.Schema;
+
+namespace Accounting;
+
+[Table("shipping")]
+[PrimaryKey(nameof(ShippingTrackingId))]
+internal class ShippingEntity
+{
+
+    public required string ShippingTrackingId { get; set; }
+
+    public required string ShippingCostCurrencyCode { get; set; }
+
+    public required long ShippingCostUnits { get; set; }
+
+    public required int ShippingCostNanos { get; set; }
+
+    public required string StreetAddress { get; set; }
+
+    public required string City { get; set; }
+
+    public required string State { get; set; }
+
+    public required string Country { get; set; }
+
+    public required string ZipCode { get; set; }
+
+    public required string OrderId { get; set; }
+}
+
+[Table("orderitem")]
+[PrimaryKey(nameof(ProductId), nameof(OrderId))]
+internal class OrderItemEntity
+{
+    public required string ItemCostCurrencyCode { get; set; }
+
+    public required long ItemCostUnits { get; set; }
+
+    public required int ItemCostNanos { get; set; }
+
+    public required string ProductId { get; set; }
+
+    public required int Quantity { get; set; }
+
+    public required string OrderId { get; set; }
+}
+
+[Table("order")]
+[PrimaryKey(nameof(Id))]
+internal class OrderEntity
+{
+    [Column("order_id")]
+    public required string Id { get; set; }
+
+}

src/accounting/README.md

@@ -7,7 +7,8 @@ This service consumes new orders from a Kafka topic.
 To build the service binary, run:
 
 ```sh
-cp pb/demo.proto src/accouting/proto/demo.proto # root context
+mkdir -p src/accounting/proto/ # root context
+cp pb/demo.proto src/accounting/proto/demo.proto # root context
 dotnet build # accounting service context
 ```
 

src/postgres/init.sql

@@ -0,0 +1,37 @@
+-- Copyright The OpenTelemetry Authors
+-- SPDX-License-Identifier: Apache-2.0
+
+CREATE USER otelu WITH PASSWORD 'otelp';
+
+
+-- Create a table
+CREATE TABLE "order" (
+    order_id TEXT PRIMARY KEY
+);
+
+CREATE TABLE shipping (
+    shipping_tracking_id TEXT PRIMARY KEY,
+    shipping_cost_currency_code TEXT NOT NULL,
+    shipping_cost_units BIGINT NOT NULL,
+    shipping_cost_nanos INT NOT NULL,
+    street_address TEXT,
+    city TEXT,
+    state TEXT,
+    country TEXT,
+    zip_code TEXT,
+    order_id TEXT NOT NULL,
+    FOREIGN KEY (order_id) REFERENCES "order"(order_id) ON DELETE CASCADE
+);
+
+CREATE TABLE orderitem (
+    item_cost_currency_code TEXT NOT NULL,
+    item_cost_units BIGINT NOT NULL,
+    item_cost_nanos INT NOT NULL,
+    product_id TEXT NOT NULL,
+    quantity INT NOT NULL,
+    order_id TEXT NOT NULL,
+    PRIMARY KEY (order_id, product_id),
+    FOREIGN KEY (order_id) REFERENCES "order"(order_id) ON DELETE CASCADE
+);
+
+GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA public TO otelu;