Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Propogate context for prometheus queries via canaryconfig ops (#2527)

Browse Source 
Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2022-08-31 12:03:00 +05:30
committed by GitHub
parent 6971bcd287
commit fef2d8f875
3 changed files with 18 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/canaryconfigmgr/canaryConfigMgr.go
+4 -4
View File
@@ -199,7 +199,7 @@ func (canaryCfgMgr *canaryConfigMgr) processCanaryConfig(ctx *context.Context, c
zap.String("name", canaryConfig.ObjectMeta.Name), zap.String("name", canaryConfig.ObjectMeta.Name),
zap.String("namespace", canaryConfig.ObjectMeta.Namespace), zap.String("namespace", canaryConfig.ObjectMeta.Namespace),
zap.String("version", canaryConfig.ObjectMeta.ResourceVersion)) zap.String("version", canaryConfig.ObjectMeta.ResourceVersion))
canaryCfgMgr.RollForwardOrBack(canaryConfig, quit, ticker) canaryCfgMgr.RollForwardOrBack(*ctx, canaryConfig, quit, ticker)
case <-quit: case <-quit:
// we're done processing this canary config either because the new function receives 100% of the traffic // we're done processing this canary config either because the new function receives 100% of the traffic
@@ -221,7 +221,7 @@ func (canaryCfgMgr *canaryConfigMgr) processCanaryConfig(ctx *context.Context, c
} }
} }
func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryConfig, quit chan struct{}, ticker *time.Ticker) { func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(ctx context.Context, canaryConfig *fv1.CanaryConfig, quit chan struct{}, ticker *time.Ticker) {
// handle race between delete event and notification on ticker.C // handle race between delete event and notification on ticker.C
_, err := canaryCfgMgr.canaryCfgCancelFuncMap.lookup(&canaryConfig.ObjectMeta) _, err := canaryCfgMgr.canaryCfgCancelFuncMap.lookup(&canaryConfig.ObjectMeta)
if err != nil { if err != nil {
@@ -233,7 +233,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC
} }
// get the http trigger object associated with this canary config // get the http trigger object associated with this canary config
triggerObj, err := canaryCfgMgr.fissionClient.CoreV1().HTTPTriggers(canaryConfig.ObjectMeta.Namespace).Get(context.TODO(), canaryConfig.Spec.Trigger, metav1.GetOptions{}) triggerObj, err := canaryCfgMgr.fissionClient.CoreV1().HTTPTriggers(canaryConfig.ObjectMeta.Namespace).Get(ctx, canaryConfig.Spec.Trigger, metav1.GetOptions{})
if err != nil { if err != nil {
// if the http trigger is not found, then give up processing this config. // if the http trigger is not found, then give up processing this config.
if k8serrors.IsNotFound(err) { if k8serrors.IsNotFound(err) {
@@ -286,7 +286,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC
methods = append(methods, triggerObj.Spec.Method) methods = append(methods, triggerObj.Spec.Method)
} }
} }
failurePercent, err := canaryCfgMgr.promClient.GetFunctionFailurePercentage(urlPath, methods, failurePercent, err := canaryCfgMgr.promClient.GetFunctionFailurePercentage(ctx, urlPath, methods,
canaryConfig.Spec.NewFunction, canaryConfig.ObjectMeta.Namespace, canaryConfig.Spec.WeightIncrementDuration) canaryConfig.Spec.NewFunction, canaryConfig.ObjectMeta.Namespace, canaryConfig.Spec.WeightIncrementDuration)
if err != nil { if err != nil {
// silently ignore. wait for next window to increment weight // silently ignore. wait for next window to increment weight
pkg/canaryconfigmgr/prometheusClient.go
+11 -11
View File
@@ -51,11 +51,11 @@ func MakePrometheusClient(logger *zap.Logger, prometheusSvc string) (*Prometheus
}, nil }, nil
} }
func (promApiClient *PrometheusApiClient) GetFunctionFailurePercentage(path string, methods []string, funcName, funcNs string, window string) (float64, error) { func (promApiClient *PrometheusApiClient) GetFunctionFailurePercentage(ctx context.Context, path string, methods []string, funcName, funcNs string, window string) (float64, error) {
var reqs, failedReqs float64 var reqs, failedReqs float64
// first get a total count of requests to this url in a time window // first get a total count of requests to this url in a time window
for _, method := range methods { for _, method := range methods {
mreqs, err := promApiClient.GetRequestsToFuncInWindow(path, method, funcName, funcNs, window) mreqs, err := promApiClient.GetRequestsToFuncInWindow(ctx, path, method, funcName, funcNs, window)
if err != nil { if err != nil {
return 0, err return 0, err
} }
@@ -68,7 +68,7 @@ func (promApiClient *PrometheusApiClient) GetFunctionFailurePercentage(path stri
// next, get a total count of errored out requests to this function in the same window // next, get a total count of errored out requests to this function in the same window
for _, method := range methods { for _, method := range methods {
mfailedReqs, err := promApiClient.GetTotalFailedRequestsToFuncInWindow(funcName, funcNs, path, method, window) mfailedReqs, err := promApiClient.GetTotalFailedRequestsToFuncInWindow(ctx, funcName, funcNs, path, method, window)
if err != nil { if err != nil {
return 0, err return 0, err
} }
@@ -85,18 +85,18 @@ func (PrometheusApiClient *PrometheusApiClient) getFunctionQueryLabels(functionN
return fmt.Sprintf("function_name=\"%s\",function_namespace=\"%s\",path=\"%s\",method=\"%s\"", functionName, functionNamespace, path, method) return fmt.Sprintf("function_name=\"%s\",function_namespace=\"%s\",path=\"%s\",method=\"%s\"", functionName, functionNamespace, path, method)
} }
func (promApiClient *PrometheusApiClient) GetRequestsToFuncInWindow(path string, method string, funcName string, funcNs string, window string) (float64, error) { func (promApiClient *PrometheusApiClient) GetRequestsToFuncInWindow(ctx context.Context, path string, method string, funcName string, funcNs string, window string) (float64, error) {
queryLabels := promApiClient.getFunctionQueryLabels(funcName, funcNs, path, method) queryLabels := promApiClient.getFunctionQueryLabels(funcName, funcNs, path, method)
queryString := fmt.Sprintf("fission_function_calls_total{%s}[%v]", queryLabels, window) queryString := fmt.Sprintf("fission_function_calls_total{%s}[%v]", queryLabels, window)
reqs, err := promApiClient.executeQuery(queryString) reqs, err := promApiClient.executeQuery(ctx, queryString)
if err != nil { if err != nil {
return 0, errors.Wrapf(err, "error executing query: %s", queryString) return 0, errors.Wrapf(err, "error executing query: %s", queryString)
} }
queryString = fmt.Sprintf("fission_function_calls_total{%s} offset %v", queryLabels, window) queryString = fmt.Sprintf("fission_function_calls_total{%s} offset %v", queryLabels, window)
reqsInPrevWindow, err := promApiClient.executeQuery(queryString) reqsInPrevWindow, err := promApiClient.executeQuery(ctx, queryString)
if err != nil { if err != nil {
return 0, errors.Wrapf(err, "error executing query: %s", queryString) return 0, errors.Wrapf(err, "error executing query: %s", queryString)
} }
@@ -111,18 +111,18 @@ func (promApiClient *PrometheusApiClient) GetRequestsToFuncInWindow(path string,
return reqsInCurrentWindow, nil return reqsInCurrentWindow, nil
} }
func (promApiClient *PrometheusApiClient) GetTotalFailedRequestsToFuncInWindow(funcName string, funcNs string, path string, method string, window string) (float64, error) { func (promApiClient *PrometheusApiClient) GetTotalFailedRequestsToFuncInWindow(ctx context.Context, funcName string, funcNs string, path string, method string, window string) (float64, error) {
queryLabels := promApiClient.getFunctionQueryLabels(funcName, funcNs, path, method) queryLabels := promApiClient.getFunctionQueryLabels(funcName, funcNs, path, method)
queryString := fmt.Sprintf("fission_function_errors_total{%s}[%v]", queryLabels, window) queryString := fmt.Sprintf("fission_function_errors_total{%s}[%v]", queryLabels, window)
failedRequests, err := promApiClient.executeQuery(queryString) failedRequests, err := promApiClient.executeQuery(ctx, queryString)
if err != nil { if err != nil {
return 0, errors.Wrapf(err, "error executing query: %s", queryString) return 0, errors.Wrapf(err, "error executing query: %s", queryString)
} }
queryString = fmt.Sprintf("fission_function_errors_total{%s} offset %v", queryLabels, window) queryString = fmt.Sprintf("fission_function_errors_total{%s} offset %v", queryLabels, window)
failedReqsInPrevWindow, err := promApiClient.executeQuery(queryString) failedReqsInPrevWindow, err := promApiClient.executeQuery(ctx, queryString)
if err != nil { if err != nil {
return 0, errors.Wrapf(err, "error executing query: %s", queryString) return 0, errors.Wrapf(err, "error executing query: %s", queryString)
} }
@@ -137,10 +137,10 @@ func (promApiClient *PrometheusApiClient) GetTotalFailedRequestsToFuncInWindow(f
return failedReqsInCurrentWindow, nil return failedReqsInCurrentWindow, nil
} }
func (promApiClient *PrometheusApiClient) executeQuery(queryString string) (float64, error) { func (promApiClient *PrometheusApiClient) executeQuery(ctx context.Context, queryString string) (float64, error) {
promApiClient.logger.Debug("executing prometheus query", zap.String("query", queryString)) promApiClient.logger.Debug("executing prometheus query", zap.String("query", queryString))
val, warn, err := promApiClient.client.Query(context.Background(), queryString, time.Now()) val, warn, err := promApiClient.client.Query(ctx, queryString, time.Now())
if err != nil { if err != nil {
return 0, errors.Wrapf(err, "error querying prometheus") return 0, errors.Wrapf(err, "error querying prometheus")
} }
pkg/controller/functionApi.go
+3 -3
View File
@@ -332,21 +332,21 @@ func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) {
} }
// get the pod with highest resource version // get the pod with highest resource version
err = getContainerLog(a.kubernetesClient, w, f, &pods[0]) err = getContainerLog(r.Context(), a.kubernetesClient, w, f, &pods[0])
if err != nil { if err != nil {
a.respondWithError(w, errors.Wrapf(err, "error getting container logs")) a.respondWithError(w, errors.Wrapf(err, "error getting container logs"))
return return
} }
} }
func getContainerLog(kubernetesClient kubernetes.Interface, w http.ResponseWriter, fn *fv1.Function, pod *apiv1.Pod) error { func getContainerLog(ctx context.Context, kubernetesClient kubernetes.Interface, w http.ResponseWriter, fn *fv1.Function, pod *apiv1.Pod) error {
seq := strings.Repeat("=", 35) seq := strings.Repeat("=", 35)
for _, container := range pod.Spec.Containers { for _, container := range pod.Spec.Containers {
podLogOpts := apiv1.PodLogOptions{Container: container.Name} // Only the env container, not fetcher podLogOpts := apiv1.PodLogOptions{Container: container.Name} // Only the env container, not fetcher
podLogsReq := kubernetesClient.CoreV1().Pods(pod.Namespace).GetLogs(pod.ObjectMeta.Name, &podLogOpts) podLogsReq := kubernetesClient.CoreV1().Pods(pod.Namespace).GetLogs(pod.ObjectMeta.Name, &podLogOpts)
podLogs, err := podLogsReq.Stream(context.Background()) podLogs, err := podLogsReq.Stream(ctx)
if err != nil { if err != nil {
return errors.Wrapf(err, "error streaming pod log") return errors.Wrapf(err, "error streaming pod log")
} }