Merge pull request #22 from FireTail-io/hotfix/err-callback-nil-deref

Hotfix/err callback nil deref

Author: rileyfiretail

README.md

@@ -135,12 +135,22 @@ The latest extension version of the publically accessible Lambda Layer published
 arn:aws:lambda:us-east-1:247286868737:layer:firetail-extension-${ARCH}-${VERSION}:1
 ```
 
-For example, for `ARCH=arm64` and `VERSION=v1-0-0` this should yield:
+For example, for `ARCH=amd64` and `VERSION=v1-0-5` this should yield:
 
 ```
-arn:aws:lambda:us-east-1:247286868737:layer:firetail-extension-x86_64-v1-0-4:1
+arn:aws:lambda:us-east-1:247286868737:layer:firetail-extension-x86_64-v1-0-5:1
 ```
 
+Regardless of how you add the Lambda Layer to your Lambda Function, you will also need to configure at least one environment variable: `FIRETAIL_API_TOKEN`. Find below a full list of the environment variables used by the Firetail Lambda Extension:
+
+| Environment Variable     | Default Value                                               | Description                                                  |
+| ------------------------ | ----------------------------------------------------------- | ------------------------------------------------------------ |
+| FIRETAIL_API_TOKEN       | None                                                        | Your API token for the Firetail Logging API. If left unset, no logs will be sent to the Firetail Logging API |
+| FIRETAIL_API_URL         | `https://api.logging.eu-west-1.prod.firetail.app/logs/bulk` | The URL of the Firetail Logging API                          |
+| FIRETAIL_EXTENSION_DEBUG | `false`                                                     | Enables debug logging from the extension if set to a value parsed as `true` by [strconv.ParseBool](https://pkg.go.dev/strconv#ParseBool) |
+| FIRETAIL_LOG_BUFFER_SIZE | `1000`                                                      | The maximum amount of logs the extension will hold in its buffer from which logs are batched and sent to Firetail |
+| FIRETAIL_MAX_BATCH_SIZE  | `100`                                                       | The maximum size of a batch of logs to be sent to the Firetail logging API in one request |
+
 
 
 #### Using The AWS CLI

logs_api_client.go

@@ -5,6 +5,8 @@ import (
 	"firetail-lambda-extension/firetail"
 	"net/http"
 	"sync"
+
+	"github.com/pkg/errors"
 )
 
 type Client struct {
@@ -45,11 +47,22 @@ func NewClient(options Options) (*Client, error) {
 	return client, nil
 }
 
-func (c *Client) ListenAndServe() error {
-	return c.httpServer.ListenAndServe()
+func (c *Client) Start(ctx context.Context) error {
+	err := c.httpServer.ListenAndServe()
+
+	if err != http.ErrServerClosed {
+		err = errors.WithMessage(err, "Log server closed unexpectedly")
+		c.errCallback(err)
+		c.Shutdown(ctx)
+	} else if err != nil {
+		c.errCallback(err)
+	}
+
+	return err
 }
 
 func (c *Client) Shutdown(ctx context.Context) error {
+	err := c.httpServer.Shutdown(ctx)
 	close(c.recordsChannel)
-	return c.httpServer.Shutdown(ctx)
+	return err
 }

logs_api_client_test.go

@@ -42,7 +42,7 @@ func TestNewClient(t *testing.T) {
 	shutdownWaitgroup.Add(1)
 	defer shutdownWaitgroup.Wait()
 	go func() {
-		err := client.ListenAndServe()
+		err := client.Start(context.Background())
 		assert.Equal(t, "http: Server closed", err.Error())
 		shutdownWaitgroup.Done()
 	}()
@@ -122,3 +122,50 @@ func TestNewClientSubscribeNoServer(t *testing.T) {
 	// This may vary depending upon the runtime
 	assert.Contains(t, err.Error(), "Err doing subscription request: Put \"http://127.0.0.1:0/2020-08-15/logs\"")
 }
+
+func TestNewClientWithInvalidLogServerAddress(t *testing.T) {
+	http.DefaultServeMux = new(http.ServeMux)
+	runtimeApiServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, `{"eventType": "SHUTDOWN"}`)
+	}))
+	defer runtimeApiServer.Close()
+
+	extensionID := "TEST_EXTENSION_ID"
+	ctx := context.Background()
+
+	t.Setenv("AWS_LAMBDA_RUNTIME_API", strings.Join(strings.Split(runtimeApiServer.URL, ":")[1:], ":")[2:])
+	logsApiClient, err := NewClient(
+		Options{
+			ExtensionID:      extensionID,
+			LogServerAddress: ":::",
+		},
+	)
+	require.Nil(t, err)
+
+	var closeErr error
+	errWaitgroup := sync.WaitGroup{}
+	errWaitgroup.Add(1)
+	go func() {
+		closeErr = logsApiClient.Start(ctx)
+		errWaitgroup.Done()
+	}()
+
+	errWaitgroup.Wait()
+	assert.Equal(t, "Log server closed unexpectedly: listen tcp: address :::: too many colons in address", closeErr.Error())
+}
+
+func TestNewClientWithNoRuntimeApi(t *testing.T) {
+	http.DefaultServeMux = new(http.ServeMux)
+
+	extensionID := "TEST_EXTENSION_ID"
+
+	t.Setenv("AWS_LAMBDA_RUNTIME_API", "127.0.0.1:0")
+	_, err := NewClient(
+		Options{
+			ExtensionID:      extensionID,
+			LogServerAddress: "127.0.0.1:0",
+		},
+	)
+	require.NotNil(t, err)
+	assert.Contains(t, err.Error(), "Err doing subscription request")
+}

logsapi/client.go

@@ -49,7 +49,7 @@ func TestRecordReceiverBatchCallbackFail(t *testing.T) {
 	shutdownWaitgroup.Add(1)
 	defer shutdownWaitgroup.Wait()
 	go func() {
-		err := client.ListenAndServe()
+		err := client.Start(context.Background())
 		assert.Equal(t, "http: Server closed", err.Error())
 		shutdownWaitgroup.Done()
 	}()
@@ -122,7 +122,7 @@ func TestRecordReceiverMaxBatchSize(t *testing.T) {
 	shutdownWaitgroup.Add(1)
 	defer shutdownWaitgroup.Wait()
 	go func() {
-		err := client.ListenAndServe()
+		err := client.Start(context.Background())
 		assert.Equal(t, "http: Server closed", err.Error())
 		shutdownWaitgroup.Done()
 	}()

logsapi/client_test.go

@@ -35,14 +35,15 @@ func main() {
 	}
 	log.Println("Registered extension, ID:", extensionClient.ExtensionID)
 
-	// Create a logsApiClient & remember to shut it down when we're done
-	logsApiClient, err := initLogsApiClient(logsapi.Options{
+	// Create a logsApiClient, start it & remember to shut it down when we're done
+	logsApiClient, err := logsapi.NewClient(logsapi.Options{
 		ExtensionID:      extensionClient.ExtensionID,
 		LogServerAddress: "sandbox:1234",
-	}, ctx)
+	})
 	if err != nil {
 		panic(err)
 	}
+	go logsApiClient.Start(ctx)
 	defer logsApiClient.Shutdown(ctx)
 
 	// awaitShutdown will block until a shutdown event is received, or the context is cancelled

logsapi/record_receiver_test.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestMain(t *testing.T) {
+	http.DefaultServeMux = new(http.ServeMux)
+	mockExtensionsApi := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, `{"eventType": "SHUTDOWN"}`)
+	}))
+	defer mockExtensionsApi.Close()
+
+	t.Setenv("AWS_LAMBDA_RUNTIME_API", strings.Join(strings.Split(mockExtensionsApi.URL, ":")[1:], ":")[2:])
+
+	main()
+}
+
+func TestMainDebug(t *testing.T) {
+	http.DefaultServeMux = new(http.ServeMux)
+	mockExtensionsApi := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, `{"eventType": "SHUTDOWN"}`)
+	}))
+	defer mockExtensionsApi.Close()
+
+	t.Setenv("AWS_LAMBDA_RUNTIME_API", strings.Join(strings.Split(mockExtensionsApi.URL, ":")[1:], ":")[2:])
+	t.Setenv("FIRETAIL_EXTENSION_DEBUG", "true")
+
+	main()
+}