Merge pull request #11 from metrico/lambda

AWS Lambda

.github/workflows/go.yml

@@ -74,6 +74,7 @@ jobs:
     - name: Compress Fluxpipe
       run: |
         strip fluxpipe-server
+        strip fluxpipe-lambda
         upx fluxpipe-server
         
     - id: tag_bump
@@ -106,6 +107,7 @@ jobs:
       with:
         release_config: |
             fluxpipe-server
+            fluxpipe-lambda
         tag_name: ${{ env.VERSION }}
         release_name: fluxpipe_${{ env.VERSION }}
         draft: false
@@ -120,7 +122,7 @@ jobs:
         username: ${{ github.actor }}
         password: ${{ secrets.GITHUB_TOKEN }}
 
-    - name: Docker Build and push
+    - name: Docker Build and push (server)
       if: github.event_name != 'pull_request'
       uses: docker/[email protected]
       with:
@@ -129,3 +131,15 @@ jobs:
         tags: |
            ghcr.io/metrico/fluxpipe:latest
            ghcr.io/metrico/fluxpipe:${{ env.VERSION }}
+           
+    - name: Docker Build and push (lambda)
+      if: github.event_name != 'pull_request'
+      uses: docker/[email protected]
+      with:
+        context: .
+        file: Dockerfile.lambda
+        push: true
+        tags: |
+           ghcr.io/metrico/fluxpipe-lambda:latest
+           ghcr.io/metrico/fluxpipe-lambda:${{ env.VERSION }}
+

Dockerfile.lambda

@@ -0,0 +1,9 @@
+FROM alpine as builder
+RUN apk add --allow-untrusted --update --no-cache curl ca-certificates
+WORKDIR /
+RUN curl -fsSL github.com/metrico/fluxpipe/releases/latest/download/fluxpipe-lambda -O && chmod +x fluxpipe-lambda
+
+FROM scratch
+COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
+COPY --from=builder /fluxpipe-lambda /fluxpipe-lambda
+CMD ["/fluxpipe-lambda"]

build.sh

@@ -3,3 +3,5 @@ export PKG_CONFIG_PATH=$(pwd)
 
 echo "Building fluxpipe-server ..."
 go build -a -ldflags '-extldflags "-static -w -ldl"' -o fluxpipe-server fluxpipe-server.go
+echo "Building fluxpipe-lambda ..."
+go build -a -ldflags '-extldflags "-static -w -ldl"' -o fluxpipe-lambda fluxpipe-lambda.go

fluxpipe-lambda.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+	"bytes"
+
+	"github.com/influxdata/flux"
+	"github.com/influxdata/flux/csv"
+	_ "github.com/influxdata/flux/fluxinit/static"
+	"github.com/influxdata/flux/lang"
+	"github.com/influxdata/flux/memory"
+	"github.com/influxdata/flux/runtime"
+
+	_fluxhttp "github.com/influxdata/flux/dependencies/http"
+	"github.com/influxdata/flux/dependencies/secret"
+	"github.com/influxdata/flux/dependencies/url"
+
+	"github.com/aws/aws-lambda-go/lambda"
+
+)
+
+var APPNAME = "fluxpipe-lambda"
+
+func runQuery(ctx context.Context, script string) (flux.Query, error) {
+
+	program, err := lang.Compile(ctx, script, runtime.Default, time.Now())
+	if err != nil {
+		return nil, err
+	}
+
+	q, err := program.Start(ctx, memory.DefaultAllocator)
+	if err != nil {
+		return nil, err
+	}
+	return q, nil
+}
+
+// NewCustomDependencies produces a Custom set of dependencies including EnvironmentSecretService.
+func NewCustomDependencies() flux.Deps {
+	validator := url.PassValidator{}
+	return flux.Deps{
+		Deps: flux.WrappedDeps{
+			HTTPClient: _fluxhttp.NewLimitedDefaultClient(validator),
+			// Default to having no filesystem, no secrets, and no url validation (always pass).
+			FilesystemService: nil,
+			SecretService:     secret.EnvironmentSecretService{},
+			URLValidator:      validator,
+		},
+	}
+}
+
+func exec(inputString string) (string, string) {
+
+	// CustomDeps produces a Custom set of dependencies including EnvironmentSecretService.
+	customValidator := url.PassValidator{}
+	customDeps := flux.Deps{
+		Deps: flux.WrappedDeps{
+			HTTPClient:        _fluxhttp.NewLimitedDefaultClient(customValidator),
+			FilesystemService: nil,
+			SecretService:     secret.EnvironmentSecretService{},
+			URLValidator:      customValidator,
+		},
+	}
+
+	// ctx := flux.NewDefaultDependencies().Inject(context.Background())
+	ctx := customDeps.Inject(context.Background())
+
+	q, err := runQuery(ctx, inputString)
+	if err != nil {
+		fmt.Println("unexpected error while creating query: %s", err)
+		return "", string(fmt.Sprintf(`{"code":"invalid","message":"%v"}`, err))
+	}
+
+	results := flux.NewResultIteratorFromQuery(q)
+	defer results.Release()
+
+	buf := bytes.NewBuffer(nil)
+	encoder := csv.NewMultiResultEncoder(csv.DefaultEncoderConfig())
+
+	if _, err := encoder.Encode(buf, results); err != nil {
+		panic(err)
+	}
+
+	q.Done()
+
+	if q.Err() != nil {
+		fmt.Println("unexpected error from query execution: %s", q.Err())
+		return "", string(fmt.Sprintf(`{"code":"invalid","message":"%v"}`, q.Err()))
+
+	} else {
+		return buf.String(), ""
+	}
+}
+
+type FluxEvent struct {
+        Query string `json:"flux"`
+}
+
+func HandleRequest(ctx context.Context, flux FluxEvent) (string, error) {
+        buf, _ := exec(flux.Query)
+        return buf, nil
+}
+
+func main() {
+        lambda.Start(HandleRequest)
+}

go.mod

@@ -3,6 +3,7 @@ module github.com/lmangani/fluxpipe
 go 1.18
 
 require (
+	github.com/aws/aws-lambda-go v1.13.3
 	github.com/influxdata/flux v0.193.0
 	github.com/labstack/echo/v4 v4.2.1
 )

go.sum

@@ -102,6 +102,7 @@ github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hC
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/aryann/difflib v0.0.0-20170710044230-e206f873d14a/go.mod h1:DAHtR1m6lCRdSC2Tm3DSWRPvIPr6xNKyeHdqDQSQT+A=
+github.com/aws/aws-lambda-go v1.13.3 h1:SuCy7H3NLyp+1Mrfp+m80jcbi9KYWAs9/BXwppwRDzY=
 github.com/aws/aws-lambda-go v1.13.3/go.mod h1:4UKl9IzQMoD+QF79YdCuzCwp8VbmG4VAQwij/eHl5CU=
 github.com/aws/aws-sdk-go v1.27.0/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.29.16/go.mod h1:1KvfttTE3SPKMpo8g2c6jL3ZKfXtFvKscTgahTma5Xg=