crossplane

1
/*
2
Copyright 2023 The Crossplane Authors.
3

4
Licensed under the Apache License, Version 2.0 (the "License"); you may not use
5
this file except in compliance with the License. You may obtain a copy of the
6
License at
7

8
    http://www.apache.org/licenses/LICENSE-2.0
9

10
Unless required by applicable law or agreed to in writing, software distributed
11
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
12
CONDITIONS OF ANY KIND, either express or implied. See the License for the
13
specific language governing permissions and limitations under the License.
14
*/
15

16
package xfn
17

18
import (
19
	"context"
20
	"time"
21

22
	"github.com/prometheus/client_golang/prometheus"
23
	"google.golang.org/grpc"
24
	"google.golang.org/grpc/status"
25

26
	"github.com/crossplane/crossplane/apis/apiextensions/fn/proto/v1beta1"
27
)
28

29
// Metrics are requests, errors, and duration (RED) metrics for composition
30
// function runs.
31
type Metrics struct {
32
	requests  *prometheus.CounterVec
33
	responses *prometheus.CounterVec
34
	duration  *prometheus.HistogramVec
35
}
36

37
// NewMetrics creates metrics for composition function runs.
38
func NewMetrics() *Metrics {
39
	return &Metrics{
40
		requests: prometheus.NewCounterVec(prometheus.CounterOpts{
41
			Subsystem: "composition",
42
			Name:      "run_function_request_total",
43
			Help:      "Total number of RunFunctionRequests sent.",
44
		}, []string{"function_name", "function_package", "grpc_target"}),
45

46
		responses: prometheus.NewCounterVec(prometheus.CounterOpts{
47
			Subsystem: "composition",
48
			Name:      "run_function_response_total",
49
			Help:      "Total number of RunFunctionResponses received.",
50
		}, []string{"function_name", "function_package", "grpc_target", "grpc_code", "result_severity"}),
51

52
		duration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
53
			Subsystem: "composition",
54
			Name:      "run_function_seconds",
55
			Help:      "Histogram of RunFunctionResponse latency (seconds).",
56
			Buckets:   prometheus.DefBuckets,
57
		}, []string{"function_name", "function_package", "grpc_target", "grpc_code", "result_severity"}),
58
	}
59
}
60

61
// Describe sends the super-set of all possible descriptors of metrics
62
// collected by this Collector to the provided channel and returns once
63
// the last descriptor has been sent.
64
func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
65
	m.requests.Describe(ch)
66
	m.responses.Describe(ch)
67
	m.duration.Describe(ch)
68
}
69

70
// Collect is called by the Prometheus registry when collecting
71
// metrics. The implementation sends each collected metric via the
72
// provided channel and returns once the last metric has been sent.
73
func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
74
	m.requests.Collect(ch)
75
	m.responses.Collect(ch)
76
	m.duration.Collect(ch)
77
}
78

79
// CreateInterceptor returns a gRPC UnaryClientInterceptor for the named
80
// function. The supplied package (pkg) should be the package's OCI reference.
81
func (m *Metrics) CreateInterceptor(name, pkg string) grpc.UnaryClientInterceptor {
82
	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
83
		l := prometheus.Labels{"function_name": name, "function_package": pkg, "grpc_target": cc.Target()}
84

85
		m.requests.With(l).Inc()
86

87
		start := time.Now()
88
		err := invoker(ctx, method, req, reply, cc, opts...)
89
		duration := time.Since(start)
90

91
		s, _ := status.FromError(err)
92
		l["grpc_code"] = s.Code().String()
93

94
		// We consider the 'severity' of the response to be that of the most
95
		// severe result in the response. A response with no results, or only
96
		// normal results, has severity "Normal". A response with warnings, but
97
		// no fatal results, has severity "Warning". A response with fatal
98
		// results has severity "Fatal".
99
		l["result_severity"] = "Normal"
100
		if rsp, ok := reply.(*v1beta1.RunFunctionResponse); ok {
101
			for _, r := range rsp.GetResults() {
102
				// Keep iterating if we see a warning result - we might still
103
				// see a fatal result.
104
				if r.GetSeverity() == v1beta1.Severity_SEVERITY_WARNING {
105
					l["result_severity"] = "Warning"
106
				}
107
				// Break if we see a fatal result, to ensure we don't downgrade
108
				// the severity to warning.
109
				if r.GetSeverity() == v1beta1.Severity_SEVERITY_FATAL {
110
					l["result_severity"] = "Fatal"
111
					break
112
				}
113
			}
114
		}
115

116
		m.responses.With(l).Inc()
117
		m.duration.With(l).Observe(duration.Seconds())
118

119
		return err
120
	}
121
}
122