Add namespace in function name for metrics #1488
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -6,10 +6,12 @@ package metrics | |
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| "io/ioutil" | ||
| "net" | ||
| "net/http" | ||
| "net/url" | ||
| "path" | ||
| "time" | ||
|
|
||
| "log" | ||
|
|
@@ -21,17 +23,19 @@ import ( | |
|
|
||
| // Exporter is a prometheus exporter | ||
| type Exporter struct { | ||
| metricOptions MetricOptions | ||
| services []types.FunctionStatus | ||
| credentials *auth.BasicAuthCredentials | ||
| FunctionNamespace string | ||
| } | ||
|
|
||
| // NewExporter creates a new exporter for the OpenFaaS gateway metrics | ||
| func NewExporter(options MetricOptions, credentials *auth.BasicAuthCredentials, namespace string) *Exporter { | ||
| return &Exporter{ | ||
| metricOptions: options, | ||
| services: []types.FunctionStatus{}, | ||
| credentials: credentials, | ||
| FunctionNamespace: namespace, | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -56,8 +60,14 @@ func (e *Exporter) Collect(ch chan<- prometheus.Metric) { | |
|
|
||
| e.metricOptions.ServiceReplicasGauge.Reset() | ||
| for _, service := range e.services { | ||
| var serviceName string | ||
| if len(service.Namespace) > 0 { | ||
| serviceName = fmt.Sprintf("%s.%s", service.Name, service.Namespace) | ||
| } else { | ||
| serviceName = service.Name | ||
| } | ||
| e.metricOptions.ServiceReplicasGauge. | ||
| WithLabelValues(serviceName). | ||
| Set(float64(service.Replicas)) | ||
| } | ||
|
|
||
|
|
@@ -72,9 +82,45 @@ func (e *Exporter) StartServiceWatcher(endpointURL url.URL, metricsOptions Metri | |
| ticker := time.NewTicker(interval) | ||
| quit := make(chan struct{}) | ||
|
|
||
| go func() { | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
|
|
||
| namespaces, err := e.getNamespaces(endpointURL) | ||
| if err != nil { | ||
| log.Println(err) | ||
| } | ||
|
|
||
| if len(namespaces) == 0 { | ||
| services, err := e.getFunctions(endpointURL, e.FunctionNamespace) | ||
| if err != nil { | ||
| log.Println(err) | ||
| continue | ||
| } | ||
| e.services = services | ||
| } else { | ||
| for _, namespace := range namespaces { | ||
| services, err := e.getFunctions(endpointURL, namespace) | ||
|
There was a problem hiding this comment. I think this works well for now, but I also suspect a larger discussion is needed about how/if we move this into the provider sending a request per namespace every second seems like something the provider should encapsulate so that it can optimize that behavior as it best makes sense for itself. This current implementation would create a "thundering herd"-like problem with no way to really streamline or optimize it. For example, the k8s provider should be able to request the existing functions from several/all namespaces in a single request. Perhaps we need a new endpoint in the provider that allows the gateway to send 0, 1, or more namespaces as part of the request. When it is 0, it would request for all namespaces, otherwise the response would send back a list of functions for the provided namespaces. Alternatively, we could allow a special There was a problem hiding this comment. I completely agree and we should move function aggregation logic to the provider. I think we can modify the exiting function list endpoint using some special query parameters like There was a problem hiding this comment. Introducing a thundering herd issue concerns me. What are the alternatives to doing that? There was a problem hiding this comment. Thundering herd issue is because every 5 second (configurable) it tries to get all namespaces and then for each namespace it lists its functions to count service/replica count of functions. Solution is to use a new endpoint or existing which list all functions across all namespaces so that we don't have to make these multiple calls. This implementation can be done in faas-netes which will be efficient. There was a problem hiding this comment. @viveksyngh can you write a bash loop to deploy 100-200 functions and then measure the effect of the patch? There was a problem hiding this comment. for j in {1..10} do ;
kubectl create namespace $i #annotate it
for i in {1..10} do ;
faas-cli deploy --image figlet --name fn$I --namespace $j
done
donePerhaps create 10-20 in 10 different namespaces |
||
| if err != nil { | ||
| log.Println(err) | ||
| continue | ||
| } | ||
| e.services = append(e.services, services...) | ||
| } | ||
| } | ||
|
|
||
| break | ||
| case <-quit: | ||
| return | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| func (e *Exporter) getHTTPClient(timeout time.Duration) http.Client { | ||
|
|
||
| return http.Client{ | ||
| Transport: &http.Transport{ | ||
| Proxy: http.ProxyFromEnvironment, | ||
| DialContext: (&net.Dialer{ | ||
|
|
@@ -87,45 +133,71 @@ func (e *Exporter) StartServiceWatcher(endpointURL url.URL, metricsOptions Metri | |
| ExpectContinueTimeout: 1500 * time.Millisecond, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func (e *Exporter) getFunctions(endpointURL url.URL, namespace string) ([]types.FunctionStatus, error) { | ||
|
There was a problem hiding this comment. Can you use the CLI here instead? |
||
| timeout := 3 * time.Second | ||
| proxyClient := e.getHTTPClient(timeout) | ||
|
|
||
| endpointURL.Path = path.Join(endpointURL.Path, "/system/functions") | ||
| if len(namespace) > 0 { | ||
| q := endpointURL.Query() | ||
| q.Set("namespace", namespace) | ||
| endpointURL.RawQuery = q.Encode() | ||
| } | ||
|
|
||
| get, _ := http.NewRequest(http.MethodGet, endpointURL.String(), nil) | ||
| if e.credentials != nil { | ||
| get.SetBasicAuth(e.credentials.User, e.credentials.Password) | ||
| } | ||
|
|
||
| services := []types.FunctionStatus{} | ||
| res, err := proxyClient.Do(get) | ||
| if err != nil { | ||
| return services, err | ||
| } | ||
|
|
||
| bytesOut, readErr := ioutil.ReadAll(res.Body) | ||
| if readErr != nil { | ||
| return services, readErr | ||
| } | ||
|
|
||
| unmarshalErr := json.Unmarshal(bytesOut, &services) | ||
| if unmarshalErr != nil { | ||
| return services, unmarshalErr | ||
| } | ||
| return services, nil | ||
| } | ||
|
|
||
| func (e *Exporter) getNamespaces(endpointURL url.URL) ([]string, error) { | ||
|
There was a problem hiding this comment. Can we use the CLI's code for this? |
||
| namespaces := []string{} | ||
| endpointURL.Path = path.Join(endpointURL.Path, "system/namespaces") | ||
|
|
||
| get, _ := http.NewRequest(http.MethodGet, endpointURL.String(), nil) | ||
| if e.credentials != nil { | ||
| get.SetBasicAuth(e.credentials.User, e.credentials.Password) | ||
| } | ||
|
|
||
| timeout := 3 * time.Second | ||
| proxyClient := e.getHTTPClient(timeout) | ||
|
|
||
| res, err := proxyClient.Do(get) | ||
| if err != nil { | ||
| return namespaces, err | ||
| } | ||
|
|
||
| if res.StatusCode == http.StatusNotFound { | ||
| return namespaces, nil | ||
| } | ||
|
|
||
| bytesOut, readErr := ioutil.ReadAll(res.Body) | ||
| if readErr != nil { | ||
| return namespaces, readErr | ||
| } | ||
|
|
||
| unmarshalErr := json.Unmarshal(bytesOut, &namespaces) | ||
| if unmarshalErr != nil { | ||
| return namespaces, unmarshalErr | ||
| } | ||
| return namespaces, nil | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What is this new segment of code for? I was mainly expecting this PR to set a default suffix in the invocation URL when there was not one given.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
New segment of code is for getting replica count of the services, which was not having support for multiple namespace. It was only getting replica count for the functions present in default namespace.