Ingest from Publisher side to track drop rate

clients.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"strconv"
+	"sync"
+)
+
+type RawMsgs struct {
+	Records []PubsubMsg `json:"records,omitempty"`
+}
+
+type KustoIngestionClient struct {
+	mutex sync.Mutex
+	requestUri string
+	pubSubMsgs []PubsubMsg
+	counter int
+	batchSize int
+}
+
+func NewKustoIngestionClient() KustoIngestionClient {
+	targetUri := os.Getenv("TARGETURI")
+	if targetUri == "" {
+		log.Fatalf("Failed to set target uri env var")
+	}
+
+	functionKey := os.Getenv("FUNCTIONKEY")
+	if functionKey == "" {
+		log.Fatalf("Failed to set function key env var")
+	}
+
+	INGESTIONSERVICEBATCHSIZE := os.Getenv("INGESTIONSERVICEBATCHSIZE")
+	batchSize, err := strconv.Atoi(INGESTIONSERVICEBATCHSIZE)
+	if INGESTIONSERVICEBATCHSIZE == "" || err != nil {
+		log.Fatalf("Failed to set INGESTIONSERVICEBATCHSIZE %v", err)
+	}
+
+	requestUri := targetUri + "?code=" + functionKey;
+
+	pubSubMsgs := make([]PubsubMsg, 0, batchSize)
+
+	return KustoIngestionClient { 
+		requestUri:requestUri,
+		pubSubMsgs: pubSubMsgs,
+		batchSize: batchSize,
+	}
+}
+
+func (client *KustoIngestionClient) SendAsync(msg PubsubMsg) error {
+	client.mutex.Lock()
+	defer client.mutex.Unlock()
+
+	client.pubSubMsgs = append(client.pubSubMsgs, msg);
+	client.counter += 1
+
+	if client.counter == client.batchSize {
+		msgs := make([]PubsubMsg, client.batchSize, client.batchSize)
+		copy(msgs, client.pubSubMsgs)
+
+		client.counter = 0
+		client.pubSubMsgs = make([]PubsubMsg, 0, client.batchSize)
+		return client.SendAsyncBatch(msgs)
+	}
+
+	return nil
+}
+
+func (client *KustoIngestionClient) SendAsyncBatch(msgs []PubsubMsg) error {
+	rawMsgs := RawMsgs{Records: msgs}
+
+	buf, err := json.Marshal(rawMsgs)
+	if err != nil {
+		log.Fatalf(err.Error())
+	}
+
+	var resp *http.Response
+	for i := 0; i < 3; i++ {
+		resp, err = http.Post(client.requestUri, "application/json", bytes.NewBuffer(buf))
+		if err == nil {
+			break
+		}
+		fmt.Printf("Error publishing count %d with error: %v \n", i, err)
+	}
+
+	if err != nil {
+		return err
+	}
+
+	fmt.Println("Response Status:", resp.Status)
+
+	return nil
+}

go.mod

@@ -6,6 +6,7 @@ require github.com/dapr/go-sdk v1.6.0
 
 require (
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	golang.org/x/net v0.0.0-20220621193019-9d032be2e588 // indirect
 	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect

go.sum

@@ -48,6 +48,7 @@ github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=

helpers.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+)
+
+func getObj(i string, batchCorrelationId string) PubsubMsg {
+	now := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
+	return PubsubMsg{
+		Key:                   i,
+		ID:                    now,
+		Details:               map[string]interface{}{"missing_labels": []interface{}{"test"}},
+		EventType:             "violation_audited",
+		Group:                 "constraints.gatekeeper.sh",
+		Version:               "v1beta1",
+		Kind:                  "K8sRequiredLabels",
+		Name:                  "pod-must-have-test",
+		Namespace:             "",
+		Message:               "you must provide labels: {\"test\"}",
+		EnforcementAction:     "deny",
+		ConstraintAnnotations: map[string]string(nil),
+		ResourceGroup:         "",
+		ResourceAPIVersion:    "v1",
+		ResourceKind:          "Pod",
+		ResourceNamespace:     "nginx",
+		ResourceName:          "dywuperf-deployment-10kpods-69bd64c867-h2wdx",
+		ResourceLabels:        map[string]string{"app": "dywuperf-app-100kpods", "pod-template-hash": "69bd64c867"},
+		Timestamp:             now,
+		BrokerName:            BROKERNAME,
+		UserAgent:             "Publisher",
+		CorrelationId:         uuid.New().String(),
+		BatchCorrelationId:    batchCorrelationId}
+}

main.go

@@ -4,15 +4,16 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log"
 	"os"
 	"strconv"
 	"time"
 
 	"github.com/dapr/go-sdk/client"
+	"github.com/google/uuid"
 )
 
-// code
 var (
 	PUBSUB_NAME = "pubsub"
 	TOPIC_NAME  = "audit"
@@ -51,6 +52,11 @@ type PubsubMsg struct {
 	// Additional Metadata for benchmarking
 	BrokerName            string            `json:"brokerName,omitempty"`
 	Timestamp             string            `json:"timestamp,omitempty"`
+	UserAgent             string            `json:"userAgent,omitempty"`
+	CorrelationId         string            `json:"correlationId,omitempty"`
+	PublishedTimestamp    string            `json:"publishedTimestamp,omitempty"`
+	ReceivedTimestamp     string            `json:"receivedTimestamp,omitempty"`
+	BatchCorrelationId    string            `json:"batchCorrelationId,omitempty"`
 }
 
 func main() {
@@ -69,58 +75,57 @@ func main() {
 	time.Sleep(300 * time.Hour)
 }
 
-func getObj(i string) interface{} {
-	now := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
-	return PubsubMsg{
-		Key:                   i,
-		ID:                    now,
-		Details:               map[string]interface{}{"missing_labels": []interface{}{"test"}},
-		EventType:             "violation_audited",
-		Group:                 "constraints.gatekeeper.sh",
-		Version:               "v1beta1",
-		Kind:                  "K8sRequiredLabels",
-		Name:                  "pod-must-have-test",
-		Namespace:             "",
-		Message:               "you must provide labels: {\"test\"}",
-		EnforcementAction:     "deny",
-		ConstraintAnnotations: map[string]string(nil),
-		ResourceGroup:         "",
-		ResourceAPIVersion:    "v1",
-		ResourceKind:          "Pod",
-		ResourceNamespace:     "nginx",
-		ResourceName:          "dywuperf-deployment-10kpods-69bd64c867-h2wdx",
-		ResourceLabels:        map[string]string{"app": "dywuperf-app-100kpods", "pod-template-hash": "69bd64c867"},
-		Timestamp:             now,
-		BrokerName:            BROKERNAME}
-}
-
 func (r *Dapr) Send() {
+	var kustoIngestionClient =  NewKustoIngestionClient()
+
 	value := os.Getenv("NUMBER")
 	log.Printf("sending %s messages", value)
 	total, err := strconv.Atoi(value)
 	if err != nil {
 		log.Fatalf("error getting number: %v", err)
-		total = 100000
+		total = 1000000
 	}
 	ctx := context.Background()
-	start_time := time.Now()
-	log.Println("starting publish")
-
-	for i := 0; i < total; i++ {
-		test := getObj(strconv.Itoa(i))
-		jsonData, err := json.Marshal(test)
-		if err != nil {
-			log.Fatalf("error marshaling data: %v", err)
-		}
 
-		for _, c := range r.client {
-			//Using Dapr SDK to publish a topic
-			// log.Println("Published data: " + string(jsonData))
-			if err := c.client.PublishEvent(ctx, PUBSUB_NAME, TOPIC_NAME, jsonData); err != nil {
-				panic(err)
+	// Always run
+	for true {
+		log.Println("starting publish")
+		start_time := time.Now()
+
+		// Each batch has a separate correlation id
+		batchCorrelationId := uuid.New().String();
+
+		for i := 0; i < total; i ++ {
+			msg := getObj(strconv.Itoa(i), batchCorrelationId)
+
+			for _, c := range r.client {
+
+				// Send to ingestion service to track drop rate
+				err = kustoIngestionClient.SendAsync(msg)
+				if err != nil {
+					// Continue if ingestion error, to do track count of these
+					log.Printf("Ingestion error for batchCorrelationId: %s, %v", batchCorrelationId, err)
+				}
+
+				// Set the published timestamp
+				msg.PublishedTimestamp = time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
+				jsonData, err := json.Marshal(msg)
+				if err != nil {
+					log.Fatalf("error marshaling data: %v", err)
+				}
+
+				// Using Dapr SDK to publish a topic and send to Subscriber
+				// What is the publisher's streaming rate? Does the batching above significantly impact this rate
+				if err := c.client.PublishEvent(ctx, PUBSUB_NAME, TOPIC_NAME, jsonData); err != nil {
+					panic(err)
+				}
 			}
 		}
+
+		end_time := time.Now()
+		log.Printf("total time it took %v", end_time.Sub(start_time))
+
+		fmt.Println("Sleep starting now: ", time.Now())
+		time.Sleep(15*time.Minute);
 	}
-	end_time := time.Now()
-	log.Printf("total time it took %v", end_time.Sub(start_time))
 }

pub.yaml

@@ -1,4 +1,4 @@
----
+# ---
 apiVersion: dapr.io/v1alpha1
 kind: Component
 metadata:
@@ -21,9 +21,9 @@ spec:
 #   name: pubsub
 #   namespace: gatekeeper-system
 # spec:
-  # type: pubsub.in-memory
-  # version: v1
-  # metadata: []
+#   type: pubsub.in-memory
+#   version: v1
+#   metadata: []
 ---
 apiVersion: apps/v1
 kind: Deployment
@@ -49,10 +49,16 @@ spec:
     spec:
       containers:
       - name: go-pub
-        image: docker.io/jaydipgabani/publisher:latest
+        image: docker.io/noreisch/test-publisher:0.0.13
         imagePullPolicy: Always
         env:
         - name: NUMBER
-          value: "1000"
+          value: "20000"
         - name: BROKERNAME
-          value: "Redis"
+          value: "Redis"
+        - name: TARGETURI
+          value: ""
+        - name: FUNCTIONKEY
+          value: ""
+        - name: INGESTIONSERVICEBATCHSIZE
+          value: "1000"