Logs dropped PR

go-example-telemetry-api-extension/Makefile

@@ -13,5 +13,5 @@ buildAndDeployExtensionLayer:
 	cd bin && zip -r extension.zip extensions
 
 	aws lambda publish-layer-version \
-		--layer-name "go-example-telemetry-api-extension-layer" \
+		--layer-name "ivan-go-example-telemetry-api-extension-layer" \
 		--zip-file "fileb://bin/extension.zip"

go-example-telemetry-api-extension/function/index.js

@@ -1,5 +1,12 @@
-console.log('Hello from function initalization');
+console.log("Hello from function initalization");
 
 exports.handler = async (event, context) => {
-    console.log('Hello from function handler', {event});
-}
+  for (let i = 0; i < 1000; i++) {
+    console.log(`I'm log ${i}`);
+  }
+  const response = {
+    statusCode: 200,
+    body: "hello, world",
+  };
+  return response;
+};

go-example-telemetry-api-extension/telemetryApi/client.go

@@ -117,14 +117,14 @@ type SubscribeResponse struct {
 func (c *Client) Subscribe(ctx context.Context, extensionId string, listenerUri string) (*SubscribeResponse, error) {
 	eventTypes := []EventType{
 		Platform,
-		// Function,
-		// Extension,
+		Function,
+		Extension,
 	}
 
 	bufferingConfig := BufferingCfg{
-		MaxItems:  1000,
-		MaxBytes:  256 * 1024,
-		TimeoutMS: 1000,
+		MaxItems:  10000,
+		MaxBytes:  1024 * 1024,
+		TimeoutMS: 30000,
 	}
 
 	destination := Destination{

go-example-telemetry-api-extension/telemetryApi/dispatcher.go

@@ -4,12 +4,12 @@
 package telemetryApi
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"net/http"
 	"os"
 	"strconv"
+	"strings"
 
 	"github.com/golang-collections/go-datastructures/queue"
 )
@@ -40,20 +40,15 @@ func NewDispatcher() *Dispatcher {
 }
 
 func (d *Dispatcher) Dispatch(ctx context.Context, logEventsQueue *queue.Queue, force bool) {
-	if !logEventsQueue.Empty() && (force || logEventsQueue.Len() >= d.minBatchSize) {
-		l.Info("[dispatcher:Dispatch] Dispatching", logEventsQueue.Len(), "log events")
-		logEntries, _ := logEventsQueue.Get(logEventsQueue.Len())
-		bodyBytes, _ := json.Marshal(logEntries)
-		req, err := http.NewRequestWithContext(ctx, "POST", d.postUri, bytes.NewBuffer(bodyBytes))
-		if err != nil {
-			panic(err)
-		}
-		_, err = d.httpClient.Do(req)
-		if err != nil {
-			l.Error("[dispatcher:Dispatch] Failed to dispatch, returning to queue:", err)
-			for logEntry := range logEntries {
-				logEventsQueue.Put(logEntry)
-			}
-		}
+	l.Info("[dispatcher:Dispatch] Dispatching", logEventsQueue.Len(), "log events")
+	logEntries, _ := logEventsQueue.Get(logEventsQueue.Len())
+	body, err := json.Marshal(logEntries)
+	if err != nil {
+		l.Error("[dispatcher:Dispatch] Failed to marshal log entries", err)
+		return
+	}
+	l.Info("[dispatcher:Dispatch] Dispatched", logEventsQueue.Len(), "log events")
+	if strings.Contains(string(body), "logsDropped") {
+		l.Info("[dispatcher:Dispatch] LOG DROPPED!")
 	}
 }

go-example-telemetry-api-extension/telemetryApi/listener.go

@@ -69,22 +69,38 @@ func (s *TelemetryApiListener) Start() (string, error) {
 // receive extension logs. Otherwise, logging here will cause Telemetry API to send new logs for
 // the printed lines which may create an infinite loop.
 func (s *TelemetryApiListener) http_handler(w http.ResponseWriter, r *http.Request) {
+
+	w.WriteHeader(200)
+
+	var requestStart = time.Now()
+	var ioReadTimeStart time.Time
+	var jsonUnmarshalTimeStart time.Time
+	var appendToQueueTimeStart time.Time
+
+	ioReadTimeStart = time.Now()
 	body, err := ioutil.ReadAll(r.Body)
 	if err != nil {
 		l.Error("[listener:http_handler] Error reading body:", err)
 		return
 	}
+	l.Infof("[time_taken] io_read_all: %v, %v bytes", time.Since(ioReadTimeStart), len(body))
 
+	jsonUnmarshalTimeStart = time.Now()
 	// Parse and put the log messages into the queue
 	var slice []interface{}
 	_ = json.Unmarshal(body, &slice)
+	l.Infof("[time_taken] json_unmarshal: %v, %v items", time.Since(jsonUnmarshalTimeStart), len(slice))
 
+	appendToQueueTimeStart = time.Now()
 	for _, el := range slice {
 		s.LogEventsQueue.Put(el)
 	}
+	l.Infof("[time_taken] append_to_queue: %v, %v items", time.Since(appendToQueueTimeStart), len(slice))
 
 	l.Info("[listener:http_handler] logEvents received:", len(slice), " LogEventsQueue length:", s.LogEventsQueue.Len())
 	slice = nil
+
+	l.Infof("[time_taken] Time taken to send HTTP OK: %v", time.Since(requestStart))
 }
 
 // Terminates the HTTP server listening for logs