MVP Demo related changes

This demo does the following:
1. Loadgenerator doesn't complete the checkout if the latency from
   adding items to the cart is more than a second. The bulk of this
   latency comes from fetching the item from the product catalog. We
   have configured the delay to be high for a specific item. This
   results in order drop by 10%, because this is one of the 10 item that
   is randomly picked.
2. Added a flag to control if productcatalog service should add a delay
   of 1 second to the item.
3. Feature flag for the "productCatalogTimeout" can be toggled on/off.

src/featureflagservice/priv/repo/migrations/20220524172636_create_featureflags.exs

@@ -23,6 +23,11 @@ defmodule Featureflagservice.Repo.Migrations.CreateFeatureflags do
       description: "Fail product catalog service on a specific product",
       enabled: false})
 
+     repo().insert(%Featureflagservice.FeatureFlags.FeatureFlag{
+      name: "productCatalogTimeout",
+      description: "Timeout product catalog service on a specific product",
+      enabled: false})
+
     repo().insert(%Featureflagservice.FeatureFlags.FeatureFlag{
       name: "recommendationCache",
       description: "Cache recommendations",
@@ -41,6 +46,7 @@ defmodule Featureflagservice.Repo.Migrations.CreateFeatureflags do
 
   defp execute_down do
     repo().delete(%Featureflagservice.FeatureFlags.FeatureFlag{name: "productCatalogFailure"})
+    repo().delete(%Featureflagservice.FeatureFlags.FeatureFlag{name: "productCatalogTimeout"})
     repo().delete(%Featureflagservice.FeatureFlags.FeatureFlag{name: "recommendationCache"})
     repo().delete(%Featureflagservice.FeatureFlags.FeatureFlag{name: "adServiceFailure"})
     repo().delete(%Featureflagservice.FeatureFlags.FeatureFlag{name: "cartServiceFailure"})

src/loadgenerator/locustfile.py

@@ -6,8 +6,10 @@
 
 import json
 import random
+import time
 import uuid
 from locust import HttpUser, task, between
+from locust.env import Environment
 
 from opentelemetry import context, baggage, trace
 from opentelemetry.metrics import set_meter_provider
@@ -110,7 +112,14 @@ def add_to_cart(self, user=""):
     def checkout(self):
         # checkout call with an item added to cart
         user = str(uuid.uuid1())
+        start_time = time.time()
+
         self.add_to_cart(user=user)
+
+        duration = time.time() - start_time
+        # skip checkout if the latency is more than a second
+        if (self.skip_checkout(duration)):
+            return
         checkout_person = random.choice(people)
         checkout_person["userId"] = user
         self.client.post("/api/checkout", json=checkout_person)
@@ -119,8 +128,16 @@ def checkout(self):
     def checkout_multi(self):
         # checkout call which adds 2-4 different items to cart before checkout
         user = str(uuid.uuid1())
+
+        start_time = time.time()
         for i in range(random.choice([2, 3, 4])):
             self.add_to_cart(user=user)
+
+        duration = time.time() - start_time
+        # skip checkout if the latency is more than a second
+        if (self.skip_checkout(duration)):
+            return 
+
         checkout_person = random.choice(people)
         checkout_person["userId"] = user
         self.client.post("/api/checkout", json=checkout_person)
@@ -129,3 +146,14 @@ def on_start(self):
         ctx = baggage.set_baggage("synthetic_request", "true")
         context.attach(ctx)
         self.index()
+
+    # Skip the checkout if the latency for adding items to the cart is beyond a threshold (default 1 secs)
+    def skip_checkout(self, duration):
+        # Check latency threshold
+        latency_threshold_secs = getattr(Environment().parsed_options, 'latency_threshold_secs', 1)
+        # Check the environment variable to skip checkout
+        skip_checkout_if_slow = getattr(Environment().parsed_options, 'skip_checkout_if_slow', True)
+        if skip_checkout_if_slow and duration > latency_threshold_secs:
+            return True
+
+

src/productcatalogservice/main.go

@@ -1,3 +1,7 @@
+/*
+ *   Copyright (c) 2023 
+ *   All rights reserved.
+ */
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 package main
@@ -205,6 +209,12 @@ func (p *productCatalog) GetProduct(ctx context.Context, req *pb.GetProductReque
 		return nil, status.Errorf(codes.Internal, msg)
 	}
 
+	if p.checkProductTimeout(ctx, req.Id) {
+		log.Info("Product timeout flag enabled, sleeping..")
+		// add delay of a second
+		time.Sleep(time.Second)
+	}
+
 	var found *pb.Product
 	for _, product := range catalog {
 		if req.Id == product.Id {
@@ -266,7 +276,33 @@ func (p *productCatalog) checkProductFailure(ctx context.Context, id string) boo
 		span.AddEvent("error", trace.WithAttributes(attribute.String("message", fmt.Sprintf("GetFlag Failed: %s", flagName))))
 		return false
 	}
+	attribute.Boolean(flagName, ffResponse.GetFlag().Enabled)
+	return ffResponse.GetFlag().Enabled
+}
+
+func (p *productCatalog) checkProductTimeout(ctx context.Context, id string) bool {
+	if id != "OLJCESPC7Z" || p.featureFlagSvcAddr == "" {
+		return false
+	}
 
+	conn, err := createClient(ctx, p.featureFlagSvcAddr)
+	if err != nil {
+		span := trace.SpanFromContext(ctx)
+		span.AddEvent("error", trace.WithAttributes(attribute.String("message", "Feature Flag Connection Failed")))
+		return false
+	}
+	defer conn.Close()
+
+	flagName := "productCatalogTimeout"
+	ffResponse, err := pb.NewFeatureFlagServiceClient(conn).GetFlag(ctx, &pb.GetFlagRequest{
+		Name: flagName,
+	})
+	if err != nil {
+		span := trace.SpanFromContext(ctx)
+		span.AddEvent("error", trace.WithAttributes(attribute.String("message", fmt.Sprintf("GetFlag Failed: %s", flagName))))
+		return false
+	}
+	attribute.Boolean(flagName, ffResponse.GetFlag().Enabled)
 	return ffResponse.GetFlag().Enabled
 }