Merge pull request #41 from triggermesh/metrics

Prometheus metrics exporter

Author: tzununbekov

go.mod

@@ -3,7 +3,16 @@ module github.com/triggermesh/aws-custom-runtime
 go 1.15
 
 require (
-	github.com/cloudevents/sdk-go/v2 v2.6.1
-	github.com/google/uuid v1.2.0
+	contrib.go.opencensus.io/exporter/prometheus v0.4.1
+	github.com/cloudevents/sdk-go/v2 v2.8.0
+	github.com/google/go-cmp v0.5.6 // indirect
+	github.com/google/uuid v1.3.0
 	github.com/kelseyhightower/envconfig v1.4.0
+	go.opencensus.io v0.23.0
+	go.uber.org/atomic v1.9.0 // indirect
+	go.uber.org/zap v1.19.1 // indirect
+	golang.org/x/lint v0.0.0-20210508222113-6edffad5e616 // indirect
+	golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9 // indirect
+	golang.org/x/tools v0.1.8 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 )

go.sum

@@ -31,6 +31,7 @@ import (
 	"github.com/kelseyhightower/envconfig"
 
 	"github.com/triggermesh/aws-custom-runtime/pkg/converter"
+	"github.com/triggermesh/aws-custom-runtime/pkg/metrics"
 	"github.com/triggermesh/aws-custom-runtime/pkg/sender"
 )
 
@@ -73,8 +74,9 @@ type Specification struct {
 }
 
 type Handler struct {
-	Sender    *sender.Sender
-	Converter converter.Converter
+	sender    *sender.Sender
+	converter converter.Converter
+	reporter  *metrics.EventProcessingStatsReporter
 
 	requestSizeLimit int64
 	functionTTL      int64
@@ -102,28 +104,41 @@ func setupEnv(internalAPIport string) error {
 }
 
 func (h *Handler) serve(w http.ResponseWriter, r *http.Request) {
+	eventTypeTag, eventSrcTag := metrics.DefaultRequestType, metrics.DefaultRequestSource
+	start := time.Now()
+	defer func() {
+		h.reporter.ReportProcessingLatency(time.Since(start), eventTypeTag, eventSrcTag)
+	}()
+
 	requestSizeLimitInBytes := h.requestSizeLimit * 1e+6
 	body, err := ioutil.ReadAll(http.MaxBytesReader(w, r.Body, requestSizeLimitInBytes))
 	if err != nil {
+		h.reporter.ReportProcessingError(false, eventTypeTag, eventSrcTag)
 		http.Error(w, err.Error(), http.StatusRequestEntityTooLarge)
 		return
 	}
 	defer r.Body.Close()
 
-	req, context, err := h.Converter.Request(body, r.Header)
+	req, context, err := h.converter.Request(body, r.Header)
 	if err != nil {
+		h.reporter.ReportProcessingError(false, eventTypeTag, eventSrcTag)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
+	eventTypeTag, eventSrcTag = metrics.CETagsFromContext(context)
+
 	result := enqueue(req, context, h.functionTTL*1e+9)
-	result.data, err = h.Converter.Response(result.data)
+	result.data, err = h.converter.Response(result.data)
 	if err != nil {
 		result.data = []byte(fmt.Sprintf("Response conversion error: %v", err))
 	}
-	if err := h.Sender.Send(result.data, result.statusCode, w); err != nil {
+	if err := h.sender.Send(result.data, result.statusCode, w); err != nil {
+		h.reporter.ReportProcessingError(false, eventTypeTag, eventSrcTag)
 		log.Printf("! %s %s %v\n", result.id, result.data, err)
+		return
 	}
+	h.reporter.ReportProcessingSuccess(eventTypeTag, eventSrcTag)
 }
 
 func enqueue(request []byte, context map[string]string, ttl int64) message {
@@ -134,7 +149,7 @@ func enqueue(request []byte, context map[string]string, ttl int64) message {
 		data:     request,
 		context:  context,
 	}
-	log.Printf("<- %s %s\n", task.id, task.data)
+	log.Printf("<- %s\n", task.id)
 
 	resultsChannel := make(chan message)
 	mutex.Lock()
@@ -158,7 +173,7 @@ func enqueue(request []byte, context map[string]string, ttl int64) message {
 	mutex.Lock()
 	delete(results, task.id)
 	mutex.Unlock()
-	log.Printf("-> %s %d %s\n", resp.id, resp.statusCode, resp.data)
+	log.Printf("-> %s %d\n", resp.id, resp.statusCode)
 	return resp
 }
 
@@ -283,10 +298,17 @@ func main() {
 		log.Fatalf("Cannot create converter: %v", err)
 	}
 
+	// start metrics reporter
+	mr, err := metrics.StatsExporter()
+	if err != nil {
+		log.Fatalf("Cannot start stats exporter: %v", err)
+	}
+
 	// setup sender
 	handler := Handler{
-		Sender:           sender.New(spec.Sink, conv.ContentType()),
-		Converter:        conv,
+		sender:           sender.New(spec.Sink, conv.ContentType()),
+		converter:        conv,
+		reporter:         mr,
 		requestSizeLimit: spec.RequestSizeLimit,
 		functionTTL:      spec.FunctionTTL,
 	}

main.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/kelseyhightower/envconfig"
 	"github.com/triggermesh/aws-custom-runtime/pkg/converter"
+	"github.com/triggermesh/aws-custom-runtime/pkg/metrics"
 	"github.com/triggermesh/aws-custom-runtime/pkg/sender"
 )
 
@@ -49,9 +50,16 @@ func TestNewTask(t *testing.T) {
 		log.Fatalf("Cannot create converter: %v", err)
 	}
 
+	// start metrics reporter
+	mr, err := metrics.StatsExporter()
+	if err != nil {
+		log.Fatalf("Cannot start stats exporter: %v", err)
+	}
+
 	handler := Handler{
-		Sender:           sender.New(s.Sink, conv.ContentType()),
-		Converter:        conv,
+		sender:           sender.New(s.Sink, conv.ContentType()),
+		converter:        conv,
+		reporter:         mr,
 		requestSizeLimit: s.RequestSizeLimit,
 		functionTTL:      s.FunctionTTL,
 	}

main_test.go

@@ -29,7 +29,11 @@ import (
 	"github.com/kelseyhightower/envconfig"
 )
 
-const contentType = "application/cloudevents+json"
+// CloudEvents request constant attributes.
+const (
+	ContentType = "application/cloudevents+json"
+	ContextKey  = "Lambda-Runtime-Cloudevents-Context"
+)
 
 // CloudEvent is a data structure required to map KLR responses to cloudevents
 type CloudEvent struct {
@@ -141,7 +145,7 @@ func (ce *CloudEvent) Request(request []byte, headers http.Header) ([]byte, map[
 	}
 
 	runtimeContext := map[string]string{
-		"Lambda-Runtime-Cloudevents-Context": string(ceContext),
+		ContextKey: string(ceContext),
 	}
 
 	return body, runtimeContext, nil
@@ -170,13 +174,14 @@ func parseStructuredCE(body []byte) ([]byte, map[string]string, error) {
 func parseBinaryCE(headers http.Header) map[string]string {
 	h := make(map[string]string)
 	for k, v := range headers {
-		if strings.HasPrefix(k, "Ce-") {
-			h[strings.ToLower(k[3:])] = strings.Join(v, ",")
+		k = strings.ToLower(k)
+		if strings.HasPrefix(k, "ce-") {
+			h[k[3:]] = strings.Join(v, ",")
 		}
 	}
 	return h
 }
 
 func (ce *CloudEvent) ContentType() string {
-	return contentType
+	return ContentType
 }

pkg/converter/cloudevents/cloudevents.go

@@ -0,0 +1,53 @@
+/*
+Copyright 2022 Triggermesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package metrics
+
+import (
+	"encoding/json"
+
+	"github.com/triggermesh/aws-custom-runtime/pkg/converter/cloudevents"
+	"go.opencensus.io/tag"
+)
+
+const (
+	typeAttr   = "type"
+	sourceAttr = "source"
+)
+
+// Default metric tags for raw requests.
+var (
+	DefaultRequestType   = tag.Insert(tagKeyEventType, "plain-http")
+	DefaultRequestSource = tag.Insert(tagKeyEventSource, "unknown")
+)
+
+// CETagsFromContext parses Lambda context and returns CloudEvents-specific
+// type and source tags.
+func CETagsFromContext(context map[string]string) (tag.Mutator, tag.Mutator) {
+	if context == nil {
+		return DefaultRequestType, DefaultRequestSource
+	}
+	ceContext, exists := context[cloudevents.ContextKey]
+	if !exists {
+		return DefaultRequestType, DefaultRequestSource
+	}
+	var attributes map[string]string
+	if err := json.Unmarshal([]byte(ceContext), &attributes); err != nil {
+		return DefaultRequestType, DefaultRequestSource
+	}
+	return tag.Insert(tagKeyEventType, attributes[typeAttr]),
+		tag.Insert(tagKeyEventSource, attributes[sourceAttr])
+}