Add performance meters for OpenSearch bulk operations and NDJson generation

Signed-off-by: Osmany Montero <osmontero@icloud.com>

Author: osmontero

plugins/events/main.go

@@ -118,6 +118,9 @@ func main() {
 }
 
 func (p *analysisServer) Analyze(event *plugins.Event, _ grpc.ServerStreamingServer[plugins.Alert]) error {
+	m := utils.NewMeter("Enqueue log for OpenSearch")
+	defer m.Elapsed("finished")
+
 	jLog, err := utils.ToString(event)
 	if err != nil {
 		return catcher.Error("cannot convert event to json", err, nil)

plugins/events/queue.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/threatwinds/go-sdk/plugins"
+	"github.com/threatwinds/go-sdk/utils"
 	"runtime"
 	"sync"
 	"time"
@@ -59,6 +60,9 @@ func startQueue() {
 	numCPU := runtime.NumCPU() * 2
 	for i := 0; i < numCPU; i++ {
 		go func() {
+			ndM := utils.NewMeter("Generate NDJson")
+			bulkM := utils.NewMeter("Send Bulk")
+
 			var ndMutex = &sync.Mutex{}
 			var nd = make([]opensearch.BulkItem, 0, 10)
 
@@ -69,6 +73,8 @@ func startQueue() {
 						continue
 					}
 
+					bulkM.Reset()
+
 					ndMutex.Lock()
 					err := opensearch.Bulk(context.Background(), nd)
 					if err != nil {
@@ -77,12 +83,16 @@ func startQueue() {
 
 					nd = make([]opensearch.BulkItem, 0, 10)
 					ndMutex.Unlock()
+
+					bulkM.Elapsed("bulk sent")
 				}
 			}()
 
 			for {
 				l := <-logs
 
+				ndM.Reset()
+
 				dataType := gjson.Get(l, "dataType").String()
 
 				id := gjson.Get(l, "id").String()
@@ -97,6 +107,8 @@ func startQueue() {
 					Action: "index",
 				})
 				ndMutex.Unlock()
+
+				ndM.Elapsed("nd json generated")
 			}
 		}()
 	}