Merge branch 'dev'

Author: TheTeaCat

agent/http.go

@@ -128,7 +128,7 @@ func (a HttpAgent) Init(agentID string) error {
 		return err
 	}
 
-	eventTypes := []logsapi.EventType{logsapi.Platform, logsapi.Function}
+	eventTypes := []logsapi.EventType{logsapi.Function}
 	bufferingCfg := logsapi.BufferingCfg{
 		MaxItems:  10000,
 		MaxBytes:  262144,

firetail/record.go

@@ -31,8 +31,9 @@ type RecordResponse struct {
 }
 
 // getLogEntryRequest returns the value for the request field of a Firetail SaaS LogEntry based upon the value of the firetail Record's Event value,
-// which is the event that invoked the lambda function. The current implementation supports APIGatewayProxyRequests and APIGatewayV2HTTPRequests.
-func (r *Record) getLogEntryRequest() (*LogEntryRequest, error) {
+// which is the event that invoked the lambda function, and the request time stated by that event value in UNIX milliseconds. The current implementation
+// supports APIGatewayProxyRequests and APIGatewayV2HTTPRequests.
+func (r *Record) getLogEntryRequest() (*LogEntryRequest, int64, error) {
 	var err error
 
 	var apiGatewayV1Request events.APIGatewayProxyRequest
@@ -58,7 +59,7 @@ func (r *Record) getLogEntryRequest() (*LogEntryRequest, error) {
 				logEntryRequest.Headers[header] = []string{value}
 			}
 		}
-		return logEntryRequest, nil
+		return logEntryRequest, apiGatewayV1Request.RequestContext.RequestTimeEpoch, nil
 	}
 	err = multierror.Append(err, apiGatewayV1RequestErr)
 
@@ -77,9 +78,9 @@ func (r *Record) getLogEntryRequest() (*LogEntryRequest, error) {
 		for header, value := range apiGatewayV2Request.Headers {
 			logEntryRequest.Headers[header] = []string{value}
 		}
-		return logEntryRequest, nil
+		return logEntryRequest, apiGatewayV2Request.RequestContext.TimeEpoch, nil
 	}
 	err = multierror.Append(err, apiGatewayV2RequestErr)
 
-	return nil, err
+	return nil, 0, err
 }

firetail/send_records_to_saas.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"time"
 
 	"github.com/hashicorp/go-multierror"
 )
@@ -20,14 +19,14 @@ func SendRecordsToSaaS(records []Record, apiUrl, apiKey string) (int, error) {
 
 	var errs error
 	for _, record := range records {
-		logEntryRequest, err := record.getLogEntryRequest()
+		logEntryRequest, requestTime, err := record.getLogEntryRequest()
 		if err != nil {
 			errs = multierror.Append(errs, fmt.Errorf("Err creating log entry request value, err: %s", err.Error()))
 			continue
 		}
 
 		logEntryBytes, err := json.Marshal(LogEntry{
-			DateCreated:   time.Now().UnixMilli(),
+			DateCreated:   requestTime,
 			ExecutionTime: record.ExecutionTime,
 			Request:       *logEntryRequest,
 			Response: LogEntryResponse{
@@ -59,16 +58,19 @@ func SendRecordsToSaaS(records []Record, apiUrl, apiKey string) (int, error) {
 
 	req.Header.Set("x-ft-api-key", apiKey)
 
+	// The execution of this request may be frozen at any time - we need to break this down so we know if the request
+	// was successfully written - if it was, should we make a second request? It risks double reporting assuming the
+	// request received a success response... 🤔
+	// TODO: investigate above.
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		// The request has now been made, so we can now say that the marshalledRecords were included in the request sent to Firetail
-		return marshalledRecords, multierror.Append(errs, err)
+		return 0, multierror.Append(errs, fmt.Errorf("Failed to make log request, err: %s", err.Error()))
 	}
 
 	var res map[string]interface{}
 	json.NewDecoder(resp.Body).Decode(&res)
 	if res["message"] != "success" {
-		return marshalledRecords, multierror.Append(errs, fmt.Errorf("got err response from firetail api: %v, req body:\n'%s'\n", res, string(reqBytes)))
+		return marshalledRecords, multierror.Append(errs, fmt.Errorf("Got err response from firetail api: %v, req body:\n'%s'\n", res, string(reqBytes)))
 	}
 
 	return marshalledRecords, errs

main.go

@@ -16,6 +16,7 @@ import (
 	"os/signal"
 	"path"
 	"strconv"
+	"sync"
 	"syscall"
 	"time"
 )
@@ -40,7 +41,7 @@ func main() {
 		debugLog("Firetail extension starting in debug mode")
 	}
 
-	// Get API url & API token from env vars
+	// Get API url, API token & buffer size from env vars
 	firetailApiToken := os.Getenv("FIRETAIL_API_TOKEN")
 	if firetailApiToken == "" {
 		log.Fatal("FIRETAIL_API_TOKEN not set")
@@ -50,9 +51,27 @@ func main() {
 		firetailApiUrl = "https://api.logging.eu-west-1.sandbox.firetail.app/logs/bulk"
 		debugLog("FIRETAIL_API_URL not set, defaulting to %s", firetailApiUrl)
 	}
+	var bufferSize int
+	bufferSizeStr := os.Getenv("FIRETAIL_LOG_BUFFER_SIZE")
+	if bufferSizeStr != "" {
+		bufferSize, err := strconv.Atoi(bufferSizeStr)
+		if err != nil {
+			log.Fatalf("FIRETAIL_LOG_BUFFER_SIZE value invalid, err: %s", err.Error())
+		}
+		if bufferSize < 1 {
+			log.Fatalf("FIRETAIL_LOG_BUFFER_SIZE is %d but must be >= 0", bufferSize)
+		}
+	} else {
+		bufferSize = 1000
+		debugLog("FIRETAIL_LOG_BUFFER_SIZE not set; defaulting to %d", bufferSize)
+	}
 
-	// Create a context with which we'll perform all our actions & make a channel to receive
-	// SIGTERM and SIGINT events & spawn a goroutine to call cancel() when we get one
+	// Create a channel down which the logsApiAgent will send events from the log API as []bytes
+	logQueue := make(chan []byte, bufferSize)
+
+	// Create a context with which we'll perform all our requests to the extensions API
+	// & make a channel to receive SIGTERM and SIGINT events & spawn a goroutine to call
+	// cancel() when we get one
 	ctx, cancel := context.WithCancel(context.Background())
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
@@ -69,73 +88,86 @@ func main() {
 		panic(err)
 	}
 
-	// Create a channel down which the logsApiAgent will send events from the log API as []bytes
-	logQueue := make(chan []byte)
+	// Create a Logs API agent
+	logsApiAgent, err := agent.NewHttpAgent(logQueue)
+	if err != nil {
+		log.Fatal(err.Error())
+	}
 
-	// Start a receiver routine for logQueue that'll run until logQueue is closed or a logsapi.RuntimeDone event is received
+	// Subscribe to the logs API. Logs start being delivered only after the subscription happens.
+	agentID := extensionClient.ExtensionID
+	err = logsApiAgent.Init(agentID)
+	if err != nil {
+		log.Fatal(err.Error())
+	}
+
+	// Start a receiver routine for logQueue that'll run until logQueue is closed
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	defer wg.Wait()
 	go func() {
+		defer wg.Done()
+		recordsBatch := []firetail.Record{}
 		for {
-			select {
-			case logBytes, open := <-logQueue:
-				if !open {
-					debugLog("Queue channel closed, logQueue recevier routine exiting...")
-					return
-				}
+			// Receive from the queue until there's nothing currently left in it
+			logQueueClosed := false
+		ReceiveLoop:
+			for {
+				select {
+				case logBytes, open := <-logQueue:
+					if !open {
+						debugLog("Queue channel closed & empty")
+						logQueueClosed = true
+						break ReceiveLoop
+					}
 
-				// Unmarshal the bytes into a LogMessages
-				var logMessages logsapi.LogMessages
-				err := json.Unmarshal([]byte(logBytes), &logMessages)
-				if err != nil {
-					debugLog("Err unmarshalling logBytes into logsapi.LogMessages, err: %s", err.Error())
-				}
+					var logMessages logsapi.LogMessages
+					err := json.Unmarshal([]byte(logBytes), &logMessages)
+					if err != nil {
+						debugLog("Err unmarshalling logBytes into logsapi.LogMessages, err: %s", err.Error())
+						continue
+					}
 
-				// Extract any firetail records from the log messages
-				firetailRecords, errs := firetail.ExtractFiretailRecords(logMessages)
-				// Log any errs, but still continue as it's possible not all failed
-				if errs != nil {
-					debugLog("Errs extracting firetail records, errs: %s", errs.Error())
-				}
-				// If there's no firetail records, then all failed or there were none, so there's nothing to do
-				if len(firetailRecords) == 0 {
-					debugLog("No firetail records extracted. Continuing...")
-					continue
+					newFiretailRecords, errs := firetail.ExtractFiretailRecords(logMessages)
+					// If there are errs, proceed as normal as it's possible not all failed to extract
+					if errs != nil {
+						debugLog("Errs extracting firetail records, errs: %s", errs.Error())
+					}
+					recordsBatch = append(recordsBatch, newFiretailRecords...)
+					debugLog("Extracted %d record(s) from logBytes; batch is now of size %d", len(newFiretailRecords), len(recordsBatch))
+					break
+
+				default:
+					// Give the other routines some to do their thang
+					time.Sleep(time.Nanosecond)
+					break ReceiveLoop
 				}
+			}
 
-				// Send the Firetail records to Firetail SaaS
-				debugLog("Sending %d record(s) to Firetail...", len(firetailRecords))
-				recordsSent, err := firetail.SendRecordsToSaaS(firetailRecords, firetailApiUrl, firetailApiToken)
+			// If the batch isn't empty, we can attempt to send it
+			if len(recordsBatch) > 0 {
+				// Debug log before & after the request, as it takes some time & it's helpful to know if execution was frozen at this time
+				debugLog("Sending %d record(s) to Firetail...", len(recordsBatch))
+				recordsSent, err := firetail.SendRecordsToSaaS(recordsBatch, firetailApiUrl, firetailApiToken)
 				debugLog("Sent %d record(s) to Firetail.", recordsSent)
 				if err != nil {
 					debugLog("Err sending record(s) to Firetail SaaS, err: %s", err.Error())
+					continue
 				}
+				// If sending the batch to Firetail was a success, we can clear out the batch!
+				debugLog("Clearing records batch...")
+				recordsBatch = []firetail.Record{}
+			}
 
-				// Check if logMessages contains a message of type logsapi.RuntimeDone - if it does, this routine needs to exit.
-				for _, logMessage := range logMessages {
-					if logMessage.Type == string(logsapi.RuntimeDone) {
-						debugLog("Found log message of type logsapi.RuntimeDone, logQueue receiver routine exiting...")
-						return
-					}
-				}
-			default:
-				time.Sleep(time.Nanosecond)
+			// If the logQueue is closed and the recordsBatch is now empty, then we can return
+			if logQueueClosed && len(recordsBatch) == 0 {
+				debugLog("logQueue closed & batch empty, logQueue receiver routine returning...")
+				return
 			}
 		}
 	}()
 
-	// Create a Logs API agent
-	logsApiAgent, err := agent.NewHttpAgent(logQueue)
-	if err != nil {
-		log.Fatal(err.Error())
-	}
-
-	// Subscribe to the logs API. Logs start being delivered only after the subscription happens.
-	agentID := extensionClient.ExtensionID
-	err = logsApiAgent.Init(agentID)
-	if err != nil {
-		log.Fatal(err.Error())
-	}
-
-	// This for loop will block until invoke or shutdown event is received or cancelled via the context
+	// This for loop will block until an invoke or shutdown event is received, or the context is cancelled
 	for {
 		select {
 		case <-ctx.Done():
@@ -151,7 +183,9 @@ func main() {
 
 			// Exit if we receive a SHUTDOWN event
 			if res.EventType == extension.Shutdown {
-				debugLog("Received extension shutdown event, exiting...")
+				debugLog("Received extension shutdown event, sleeping for 500ms to allow final logs to arrive...")
+				time.Sleep(500)
+				debugLog("Exiting...")
 				logsApiAgent.Shutdown()
 				close(logQueue)
 				return