8 files changed, 558 insertions, 0 deletions
diff --git a/src/cart/src/.dockerignore b/src/cart/src/.dockerignore
new file mode 100644
index 0000000..0224086
--- /dev/null
+++ b/src/cart/src/.dockerignore
@@ -0,0 +1,6 @@
+**/*.sh
+**/*.bat
+**/bin/
+**/obj/
+**/out/
+Dockerfile*
+\ No newline at end of file
diff --git a/src/cart/src/Dockerfile b/src/cart/src/Dockerfile
new file mode 100644
index 0000000..9e4df98
--- /dev/null
+++ b/src/cart/src/Dockerfile
@@ -0,0 +1,41 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+# Copyright 2021 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.
+
+# https://mcr.microsoft.com/v2/dotnet/sdk/tags/list
+FROM --platform=$BUILDPLATFORM mcr.microsoft.com/dotnet/sdk:8.0 AS builder
+ARG TARGETARCH
+
+WORKDIR /usr/src/app/
+
+COPY ./src/cart/ ./
+COPY ./pb/ ./pb/
+
+RUN dotnet restore ./src/cart.csproj -r linux-musl-$TARGETARCH
+
+RUN dotnet publish ./src/cart.csproj -r linux-musl-$TARGETARCH --no-restore -o /cart
+
+# -----------------------------------------------------------------------------
+
+# https://mcr.microsoft.com/v2/dotnet/runtime-deps/tags/list
+FROM mcr.microsoft.com/dotnet/runtime-deps:8.0-alpine3.20
+
+WORKDIR /usr/src/app/
+COPY --from=builder /cart/ ./
+
+ENV DOTNET_HOSTBUILDER__RELOADCONFIGONCHANGE=false
+
+EXPOSE ${CART_PORT}
+ENTRYPOINT [ "./cart" ]
diff --git a/src/cart/src/Program.cs b/src/cart/src/Program.cs
new file mode 100644
index 0000000..588bcb2
--- /dev/null
+++ b/src/cart/src/Program.cs
@@ -0,0 +1,99 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+using System;
+
+using cart.cartstore;
+using cart.services;
+
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+using Microsoft.Extensions.Logging;
+using OpenTelemetry.Instrumentation.StackExchangeRedis;
+using OpenTelemetry.Logs;
+using OpenTelemetry.Metrics;
+using OpenTelemetry.Resources;
+using OpenTelemetry.Trace;
+using OpenFeature;
+using OpenFeature.Contrib.Providers.Flagd;
+using OpenFeature.Hooks;
+
+var builder = WebApplication.CreateBuilder(args);
+string valkeyAddress = builder.Configuration["VALKEY_ADDR"];
+if (string.IsNullOrEmpty(valkeyAddress))
+{
+    Console.WriteLine("VALKEY_ADDR environment variable is required.");
+    Environment.Exit(1);
+}
+
+builder.Logging
+    .AddOpenTelemetry(options => options.AddOtlpExporter())
+    .AddConsole();
+
+builder.Services.AddSingleton<ICartStore>(x =>
+{
+    var store = new ValkeyCartStore(x.GetRequiredService<ILogger<ValkeyCartStore>>(), valkeyAddress);
+    store.Initialize();
+    return store;
+});
+
+builder.Services.AddOpenFeature(openFeatureBuilder =>
+{
+    openFeatureBuilder
+        .AddHostedFeatureLifecycle()
+        .AddProvider(_ => new FlagdProvider())
+        .AddHook<MetricsHook>()
+        .AddHook<TraceEnricherHook>();
+});
+
+builder.Services.AddSingleton(x =>
+    new CartService(
+        x.GetRequiredService<ICartStore>(),
+        new ValkeyCartStore(x.GetRequiredService<ILogger<ValkeyCartStore>>(), "badhost:1234"),
+        x.GetRequiredService<IFeatureClient>()
+));
+
+
+Action<ResourceBuilder> appResourceBuilder =
+    resource => resource
+        .AddService(builder.Environment.ApplicationName)
+        .AddContainerDetector()
+        .AddHostDetector();
+
+builder.Services.AddOpenTelemetry()
+    .ConfigureResource(appResourceBuilder)
+    .WithTracing(tracerBuilder => tracerBuilder
+        .AddSource("OpenTelemetry.Demo.Cart")
+        .AddRedisInstrumentation(
+            options => options.SetVerboseDatabaseStatements = true)
+        .AddAspNetCoreInstrumentation()
+        .AddGrpcClientInstrumentation()
+        .AddHttpClientInstrumentation()
+        .AddOtlpExporter())
+    .WithMetrics(meterBuilder => meterBuilder
+        .AddMeter("OpenTelemetry.Demo.Cart")
+        .AddMeter("OpenFeature")
+        .AddProcessInstrumentation()
+        .AddRuntimeInstrumentation()
+        .AddAspNetCoreInstrumentation()
+        .SetExemplarFilter(ExemplarFilterType.TraceBased)
+        .AddOtlpExporter());
+builder.Services.AddGrpc();
+builder.Services.AddGrpcHealthChecks()
+    .AddCheck("Sample", () => HealthCheckResult.Healthy());
+
+var app = builder.Build();
+
+var ValkeyCartStore = (ValkeyCartStore)app.Services.GetRequiredService<ICartStore>();
+app.Services.GetRequiredService<StackExchangeRedisInstrumentation>().AddConnection(ValkeyCartStore.GetConnection());
+
+app.MapGrpcService<CartService>();
+app.MapGrpcHealthChecksService();
+
+app.MapGet("/", async context =>
+{
+    await context.Response.WriteAsync("Communication with gRPC endpoints must be made through a gRPC client. To learn how to create a client, visit: https://go.microsoft.com/fwlink/?linkid=2086909");
+});
+
+app.Run();
diff --git a/src/cart/src/appsettings.json b/src/cart/src/appsettings.json
new file mode 100644
index 0000000..db76fce
--- /dev/null
+++ b/src/cart/src/appsettings.json
@@ -0,0 +1,15 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft": "Warning",
+      "Microsoft.Hosting.Lifetime": "Information"
+    }
+  },
+  "AllowedHosts": "*",
+  "Kestrel": {
+    "EndpointDefaults": {
+      "Protocols": "Http2"
+    }
+  }
+}
+\ No newline at end of file
diff --git a/src/cart/src/cart.csproj b/src/cart/src/cart.csproj
new file mode 100644
index 0000000..4092cc1
--- /dev/null
+++ b/src/cart/src/cart.csproj
@@ -0,0 +1,41 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ProduceReferenceAssembly>false</ProduceReferenceAssembly>
+    <StaticWebAssetsEnabled>false</StaticWebAssetsEnabled>
+    <PublishSingleFile>true</PublishSingleFile>
+    <SelfContained>true</SelfContained>
+    <PublishTrimmed>false</PublishTrimmed>
+    <ProtosDir>$(ProjectDir)..\pb</ProtosDir>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="!Exists('$(ProtosDir)')">
+    <ProtosDir>..\..\..\pb</ProtosDir>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <!-- Keeping Grpc.AspNetCore* to 2.67 due to https://github.com/grpc/grpc/issues/38538 -->
+    <PackageReference Include="Grpc.AspNetCore" Version="2.67.0" />
+    <PackageReference Include="Grpc.AspNetCore.HealthChecks" Version="2.67.0" />
+    <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.11.2" />
+    <PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="1.11.2" />
+    <PackageReference Include="OpenTelemetry.Instrumentation.AspNetCore" Version="1.11.1" />
+    <PackageReference Include="OpenTelemetry.Instrumentation.GrpcNetClient" Version="1.11.0-beta.2" />
+    <PackageReference Include="OpenTelemetry.Instrumentation.Http" Version="1.11.1" />
+    <PackageReference Include="OpenTelemetry.Instrumentation.Process" Version="1.11.0-beta.2" />
+    <PackageReference Include="OpenTelemetry.Instrumentation.StackExchangeRedis" Version="1.11.0-beta.2" />
+    <PackageReference Include="OpenTelemetry.Instrumentation.Runtime" Version="1.11.1" />
+    <PackageReference Include="OpenTelemetry.Resources.Container" Version="1.11.0-beta.2" />
+    <PackageReference Include="OpenTelemetry.Resources.Host" Version="1.11.0-beta.2" />
+    <PackageReference Include="StackExchange.Redis" Version="2.8.31" />
+    <PackageReference Include="OpenFeature.Contrib.Providers.Flagd" Version="0.3.2" />
+    <PackageReference Include="OpenFeature" Version="2.7.0" />
+    <PackageReference Include="OpenFeature.DependencyInjection" Version="2.7.0" />
+    <PackageReference Include="OpenFeature.Hosting" Version="2.7.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <Protobuf Include="$(ProtosDir)\**\*.proto" GrpcServices="Both" />
+  </ItemGroup>
+</Project>
diff --git a/src/cart/src/cartstore/ICartStore.cs b/src/cart/src/cartstore/ICartStore.cs
new file mode 100644
index 0000000..80e249e
--- /dev/null
+++ b/src/cart/src/cartstore/ICartStore.cs
@@ -0,0 +1,17 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+using System.Threading.Tasks;
+
+namespace cart.cartstore;
+
+public interface ICartStore
+{
+    void Initialize();
+
+    Task AddItemAsync(string userId, string productId, int quantity);
+    Task EmptyCartAsync(string userId);
+
+    Task<Oteldemo.Cart> GetCartAsync(string userId);
+
+    bool Ping();
+}
diff --git a/src/cart/src/cartstore/ValkeyCartStore.cs b/src/cart/src/cartstore/ValkeyCartStore.cs
new file mode 100644
index 0000000..8b230ba
--- /dev/null
+++ b/src/cart/src/cartstore/ValkeyCartStore.cs
@@ -0,0 +1,238 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Grpc.Core;
+using StackExchange.Redis;
+using Google.Protobuf;
+using Microsoft.Extensions.Logging;
+using System.Diagnostics.Metrics;
+using System.Diagnostics;
+
+namespace cart.cartstore;
+
+public class ValkeyCartStore : ICartStore
+{
+    private readonly ILogger _logger;
+    private const string CartFieldName = "cart";
+    private const int RedisRetryNumber = 30;
+
+    private volatile ConnectionMultiplexer _redis;
+    private volatile bool _isRedisConnectionOpened;
+
+    private readonly object _locker = new();
+    private readonly byte[] _emptyCartBytes;
+    private readonly string _connectionString;
+
+    private static readonly ActivitySource CartActivitySource = new("OpenTelemetry.Demo.Cart");
+    private static readonly Meter CartMeter = new Meter("OpenTelemetry.Demo.Cart");
+    private static readonly Histogram<double> addItemHistogram = CartMeter.CreateHistogram(
+        "app.cart.add_item.latency",
+        unit: "s",
+        advice: new InstrumentAdvice<double>
+        {
+            HistogramBucketBoundaries = [ 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10 ]
+        });
+    private static readonly Histogram<double> getCartHistogram = CartMeter.CreateHistogram(
+        "app.cart.get_cart.latency",
+        unit: "s",
+        advice: new InstrumentAdvice<double>
+        {
+            HistogramBucketBoundaries = [ 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10 ]
+        });
+    private readonly ConfigurationOptions _redisConnectionOptions;
+
+    public ValkeyCartStore(ILogger<ValkeyCartStore> logger, string valkeyAddress)
+    {
+        _logger = logger;
+        // Serialize empty cart into byte array.
+        var cart = new Oteldemo.Cart();
+        _emptyCartBytes = cart.ToByteArray();
+        _connectionString = $"{valkeyAddress},ssl=false,allowAdmin=true,abortConnect=false";
+
+        _redisConnectionOptions = ConfigurationOptions.Parse(_connectionString);
+
+        // Try to reconnect multiple times if the first retry fails.
+        _redisConnectionOptions.ConnectRetry = RedisRetryNumber;
+        _redisConnectionOptions.ReconnectRetryPolicy = new ExponentialRetry(1000);
+
+        _redisConnectionOptions.KeepAlive = 180;
+    }
+
+    public ConnectionMultiplexer GetConnection()
+    {
+        EnsureRedisConnected();
+        return _redis;
+    }
+
+    public void Initialize()
+    {
+        EnsureRedisConnected();
+    }
+
+    private void EnsureRedisConnected()
+    {
+        if (_isRedisConnectionOpened)
+        {
+            return;
+        }
+
+        // Connection is closed or failed - open a new one but only at the first thread
+        lock (_locker)
+        {
+            if (_isRedisConnectionOpened)
+            {
+                return;
+            }
+
+            _logger.LogDebug("Connecting to Redis: {_connectionString}", _connectionString);
+            _redis = ConnectionMultiplexer.Connect(_redisConnectionOptions);
+
+            if (_redis == null || !_redis.IsConnected)
+            {
+                _logger.LogError("Wasn't able to connect to redis");
+
+                // We weren't able to connect to Redis despite some retries with exponential backoff.
+                throw new ApplicationException("Wasn't able to connect to redis");
+            }
+
+            _logger.LogInformation("Successfully connected to Redis");
+            var cache = _redis.GetDatabase();
+
+            _logger.LogDebug("Performing small test");
+            cache.StringSet("cart", "OK" );
+            object res = cache.StringGet("cart");
+            _logger.LogDebug("Small test result: {res}", res);
+
+            _redis.InternalError += (_, e) => { Console.WriteLine(e.Exception); };
+            _redis.ConnectionRestored += (_, _) =>
+            {
+                _isRedisConnectionOpened = true;
+                _logger.LogInformation("Connection to redis was restored successfully.");
+            };
+            _redis.ConnectionFailed += (_, _) =>
+            {
+                _logger.LogInformation("Connection failed. Disposing the object");
+                _isRedisConnectionOpened = false;
+            };
+
+            _isRedisConnectionOpened = true;
+        }
+    }
+
+    public async Task AddItemAsync(string userId, string productId, int quantity)
+    {
+        var stopwatch = Stopwatch.StartNew();
+        _logger.LogInformation($"AddItemAsync called with userId={userId}, productId={productId}, quantity={quantity}");
+
+        try
+        {
+            EnsureRedisConnected();
+
+            var db = _redis.GetDatabase();
+
+            // Access the cart from the cache
+            var value = await db.HashGetAsync(userId, CartFieldName);
+
+            Oteldemo.Cart cart;
+            if (value.IsNull)
+            {
+                cart = new Oteldemo.Cart
+                {
+                    UserId = userId
+                };
+                cart.Items.Add(new Oteldemo.CartItem { ProductId = productId, Quantity = quantity });
+            }
+            else
+            {
+                cart = Oteldemo.Cart.Parser.ParseFrom(value);
+                var existingItem = cart.Items.SingleOrDefault(i => i.ProductId == productId);
+                if (existingItem == null)
+                {
+                    cart.Items.Add(new Oteldemo.CartItem { ProductId = productId, Quantity = quantity });
+                }
+                else
+                {
+                    existingItem.Quantity += quantity;
+                }
+            }
+
+            await db.HashSetAsync(userId, new[]{ new HashEntry(CartFieldName, cart.ToByteArray()) });
+            await db.KeyExpireAsync(userId, TimeSpan.FromMinutes(60));
+        }
+        catch (Exception ex)
+        {
+            throw new RpcException(new Status(StatusCode.FailedPrecondition, $"Can't access cart storage. {ex}"));
+        }
+        finally
+        {
+            addItemHistogram.Record(stopwatch.Elapsed.TotalSeconds);
+        }
+    }
+
+    public async Task EmptyCartAsync(string userId)
+    {
+        _logger.LogInformation($"EmptyCartAsync called with userId={userId}");
+
+        try
+        {
+            EnsureRedisConnected();
+            var db = _redis.GetDatabase();
+
+            // Update the cache with empty cart for given user
+            await db.HashSetAsync(userId, new[] { new HashEntry(CartFieldName, _emptyCartBytes) });
+            await db.KeyExpireAsync(userId, TimeSpan.FromMinutes(60));
+        }
+        catch (Exception ex)
+        {
+            throw new RpcException(new Status(StatusCode.FailedPrecondition, $"Can't access cart storage. {ex}"));
+        }
+    }
+
+    public async Task<Oteldemo.Cart> GetCartAsync(string userId)
+    {
+        var stopwatch = Stopwatch.StartNew();
+        _logger.LogInformation($"GetCartAsync called with userId={userId}");
+
+        try
+        {
+            EnsureRedisConnected();
+
+            var db = _redis.GetDatabase();
+
+            // Access the cart from the cache
+            var value = await db.HashGetAsync(userId, CartFieldName);
+
+            if (!value.IsNull)
+            {
+                return Oteldemo.Cart.Parser.ParseFrom(value);
+            }
+
+            // We decided to return empty cart in cases when user wasn't in the cache before
+            return new Oteldemo.Cart();
+        }
+        catch (Exception ex)
+        {
+            throw new RpcException(new Status(StatusCode.FailedPrecondition, $"Can't access cart storage. {ex}"));
+        }
+        finally
+        {
+            getCartHistogram.Record(stopwatch.Elapsed.TotalSeconds);
+        }
+    }
+
+    public bool Ping()
+    {
+        try
+        {
+            var cache = _redis.GetDatabase();
+            var res = cache.Ping();
+            return res != TimeSpan.Zero;
+        }
+        catch (Exception)
+        {
+            return false;
+        }
+    }
+}
diff --git a/src/cart/src/services/CartService.cs b/src/cart/src/services/CartService.cs
new file mode 100644
index 0000000..5578f45
--- /dev/null
+++ b/src/cart/src/services/CartService.cs
@@ -0,0 +1,101 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+using System.Diagnostics;
+using System.Threading.Tasks;
+using System;
+using Grpc.Core;
+using cart.cartstore;
+using OpenFeature;
+using Oteldemo;
+
+namespace cart.services;
+
+public class CartService : Oteldemo.CartService.CartServiceBase
+{
+    private static readonly Empty Empty = new();
+    private readonly Random random = new Random();
+    private readonly ICartStore _badCartStore;
+    private readonly ICartStore _cartStore;
+    private readonly IFeatureClient _featureFlagHelper;
+
+    public CartService(ICartStore cartStore, ICartStore badCartStore, IFeatureClient featureFlagService)
+    {
+        _badCartStore = badCartStore;
+        _cartStore = cartStore;
+        _featureFlagHelper = featureFlagService;
+    }
+
+    public override async Task<Empty> AddItem(AddItemRequest request, ServerCallContext context)
+    {
+        var activity = Activity.Current;
+        activity?.SetTag("app.user.id", request.UserId);
+        activity?.SetTag("app.product.id", request.Item.ProductId);
+        activity?.SetTag("app.product.quantity", request.Item.Quantity);
+
+        try
+        {
+            await _cartStore.AddItemAsync(request.UserId, request.Item.ProductId, request.Item.Quantity);
+
+            return Empty;
+        }
+        catch (RpcException ex)
+        {
+            activity?.AddException(ex);
+            activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
+            throw;
+        }
+    }
+
+    public override async Task<Cart> GetCart(GetCartRequest request, ServerCallContext context)
+    {
+        var activity = Activity.Current;
+        activity?.SetTag("app.user.id", request.UserId);
+        activity?.AddEvent(new("Fetch cart"));
+
+        try
+        {
+            var cart = await _cartStore.GetCartAsync(request.UserId);
+            var totalCart = 0;
+            foreach (var item in cart.Items)
+            {
+                totalCart += item.Quantity;
+            }
+            activity?.SetTag("app.cart.items.count", totalCart);
+
+            return cart;
+        }
+        catch (RpcException ex)
+        {
+            activity?.AddException(ex);
+            activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
+            throw;
+        }
+    }
+
+    public override async Task<Empty> EmptyCart(EmptyCartRequest request, ServerCallContext context)
+    {
+        var activity = Activity.Current;
+        activity?.SetTag("app.user.id", request.UserId);
+        activity?.AddEvent(new("Empty cart"));
+
+        try
+        {
+            if (await _featureFlagHelper.GetBooleanValueAsync("cartFailure", false))
+            {
+                await _badCartStore.EmptyCartAsync(request.UserId);
+            }
+            else
+            {
+                await _cartStore.EmptyCartAsync(request.UserId);
+            }
+        }
+        catch (RpcException ex)
+        {
+            Activity.Current?.AddException(ex);
+            Activity.Current?.SetStatus(ActivityStatusCode.Error, ex.Message);
+            throw;
+        }
+
+        return Empty;
+    }
+}