Merge pull request #20 from ispiroglu/unnecessary-buffers

Unnecessary buffers

.gitignore

@@ -13,4 +13,10 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
- /.idea
+ /.idea
+.idea
+.scannerwork
+
+/.vscode
+.nvim
+.nvimlog

cmd/mercurius-client/publisher-client/mercurius-client.go

@@ -19,7 +19,8 @@ const CLIENT_NAME = "Sample Client"
 
 var logger = logger2.NewLogger()
 var messageCount = atomic.Uint64{}
-var N = 100
+var N = 100 * 100
+var start time.Time
 
 func main() {
 	c, err := client.NewClient(CLIENT_NAME, ADDR)
@@ -33,20 +34,28 @@ func main() {
 	wg.Add(N)
 	for i := 0; i < N; i++ {
 		go func(w *sync.WaitGroup) {
-			for j := 0; j < 100; j++ {
-				if err := c.Publish(TopicName, []byte(strconv.FormatUint(messageCount.Load(), 10)), context.Background()); err != nil {
+			for j := 0; j < 1; j++ {
+				x := messageCount.Add(1)
+				if err := c.Publish(TopicName, []byte(strconv.FormatUint(x, 10)), context.Background()); err != nil {
 					logger.Error("Err", zap.Error(err))
 				}
-				fmt.Println(strconv.FormatUint(messageCount.Load(), 10))
-
-				messageCount.Add(1)
+				if x == 1 {
+					start = time.Now()
+				}
+				fmt.Println(x)
+				if x == 1000*1000 {
+					z := time.Since(start)
+					fmt.Println("Execution time: ", z)
+				}
+				fmt.Println(strconv.FormatUint(x, 10))
+				//time.Sleep(time.Millisecond)
 			}
 			w.Done()
 		}(&wg)
+		//time.Sleep(200 * time.Second)
 	}
 
 	wg.Wait()
-	time.Sleep(1 * time.Hour)
 }
 
 // package main

cmd/mercurius-client/subscriber-client/mercurius-client-sub.go

@@ -18,9 +18,10 @@ const CLIENT_NAME = "Sample Client"
 const N = 100 * 100 * 100
 
 var messageCount = atomic.Uint64{}
-var start time.Time = time.Time{}
+var start = time.Time{}
 var logger = k.NewLogger()
-var ctx, cancel = context.WithCancel(context.Background())
+var ctx, _ = context.WithCancel(context.Background())
+var ch = make(chan struct{})
 
 func main() {
 	c, err := client.NewClient(CLIENT_NAME, ADDR)
@@ -33,29 +34,22 @@ func main() {
 				logger.Error("Err", zap.Error(err))
 			}
 		}()
-	}
-
-	timer := time.NewTimer(900 * time.Second)
-ConsumerLoop:
-	for {
-		select {
-		case <-timer.C:
-			cancel()
-			break ConsumerLoop
 
-		}
 	}
 
+	<-ch
 }
 
 func handler(e *proto.Event) error {
-	messageCount.Add(1)
-	if messageCount.Load() == 1 {
+	x := messageCount.Add(1)
+	if x == 1 {
 		start = time.Now()
 	}
-	if messageCount.Load() == N {
+	fmt.Println(x)
+	if x == N {
 		z := time.Since(start)
 		fmt.Println("Execution time: ", z)
+		ch <- struct{}{}
 	}
 	return nil
 }

cmd/mercurius/mercurius.go

@@ -4,11 +4,12 @@ import (
 	"net"
 
 	"github.com/ispiroglu/mercurius/internal/logger"
-	"go.uber.org/zap"
-
 	sv "github.com/ispiroglu/mercurius/internal/server"
 	"github.com/ispiroglu/mercurius/proto"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
+	"net/http"
 )
 
 const ADDR = "0.0.0.0:9000"
@@ -28,11 +29,10 @@ func main() {
 
 	proto.RegisterMercuriusServer(grpcServer, server)
 
-	// go func() {
-	// 	time.Sleep(10 * time.Second)
-	// 	grpcServer.GracefulStop()
-	// }()
-
+	go func() {
+		http.Handle("/metrics", promhttp.Handler())
+		_ = http.ListenAndServe(":8081", nil)
+	}()
 	if err := grpcServer.Serve(list); err != nil {
 		log.Fatal("Failed to serve", zap.Error(err))
 	}

docker-compose.yaml

@@ -0,0 +1,37 @@
+version: "3"
+services:
+
+#  tempo:
+#    image: grafana/tempo:latest
+#    command: [ "-config.file=/etc/tempo.yaml" ]
+#    volumes:
+#      - ./prometheus/tempo.yaml:/etc/tempo.yaml
+#    ports:
+#      - "3200:3200"
+#      - "4318:4318"
+#  loki:
+#      image: grafana/loki:latest
+#      ports:
+#        - "3100:3100"
+#      command: -config.file=/etc/loki/local-config.yaml
+  prometheus:
+    image: prom/prometheus
+    ports:
+      - 9090:9090
+    command:
+      - --web.enable-remote-write-receiver
+      - --enable-feature=exemplar-storage
+    volumes:
+      - ./prometheus/prometheus.yml:/prometheus/prometheus.yml
+
+  grafana:
+    image: grafana/grafana:latest
+ #   volumes:
+  #    - ./prometheus/grafana-datasources.yaml:/etc/grafana/provisioning/datasources/datasources.yaml
+    environment:
+      - GF_AUTH_ANONYMOUS_ENABLED=true
+      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
+      - GF_AUTH_DISABLE_LOGIN_FORM=true
+      - GF_FEATURE_TOGGLES_ENABLE=traceqlEditor
+    ports:
+      - "3000:3000"

go.mod

@@ -4,21 +4,30 @@ go 1.20
 
 require (
 	github.com/google/uuid v1.3.0
+	github.com/prometheus/client_golang v1.16.0
 	github.com/stretchr/testify v1.8.4
 	go.uber.org/zap v1.24.0
 	google.golang.org/grpc v1.53.0
-	google.golang.org/protobuf v1.29.0
+	google.golang.org/protobuf v1.30.0
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.42.0 // indirect
+	github.com/prometheus/procfs v0.10.1 // indirect
+	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
-	golang.org/x/net v0.5.0 // indirect
-	golang.org/x/sys v0.4.0 // indirect
-	golang.org/x/text v0.6.0 // indirect
+	golang.org/x/net v0.7.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/text v0.7.0 // indirect
 	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )