diff --git a/pkg/canaryconfigmgr/canaryConfigMgr.go b/pkg/canaryconfigmgr/canaryConfigMgr.go index 6975c63f..3c5289b5 100644 --- a/pkg/canaryconfigmgr/canaryConfigMgr.go +++ b/pkg/canaryconfigmgr/canaryConfigMgr.go @@ -199,7 +199,7 @@ func (canaryCfgMgr *canaryConfigMgr) processCanaryConfig(ctx *context.Context, c zap.String("name", canaryConfig.ObjectMeta.Name), zap.String("namespace", canaryConfig.ObjectMeta.Namespace), zap.String("version", canaryConfig.ObjectMeta.ResourceVersion)) - canaryCfgMgr.RollForwardOrBack(canaryConfig, quit, ticker) + canaryCfgMgr.RollForwardOrBack(*ctx, canaryConfig, quit, ticker) case <-quit: // 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 _, err := canaryCfgMgr.canaryCfgCancelFuncMap.lookup(&canaryConfig.ObjectMeta) if err != nil { @@ -233,7 +233,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC } // 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 the http trigger is not found, then give up processing this config. if k8serrors.IsNotFound(err) { @@ -286,7 +286,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC 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) if err != nil { // silently ignore. wait for next window to increment weight diff --git a/pkg/canaryconfigmgr/prometheusClient.go b/pkg/canaryconfigmgr/prometheusClient.go index 79524ba0..7dbc3bb8 100644 --- a/pkg/canaryconfigmgr/prometheusClient.go +++ b/pkg/canaryconfigmgr/prometheusClient.go @@ -51,11 +51,11 @@ func MakePrometheusClient(logger *zap.Logger, prometheusSvc string) (*Prometheus }, 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 // first get a total count of requests to this url in a time window 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 { 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 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 { 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) } -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) 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 { return 0, errors.Wrapf(err, "error executing query: %s", queryString) } 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 { return 0, errors.Wrapf(err, "error executing query: %s", queryString) } @@ -111,18 +111,18 @@ func (promApiClient *PrometheusApiClient) GetRequestsToFuncInWindow(path string, 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) 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 { return 0, errors.Wrapf(err, "error executing query: %s", queryString) } 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 { return 0, errors.Wrapf(err, "error executing query: %s", queryString) } @@ -137,10 +137,10 @@ func (promApiClient *PrometheusApiClient) GetTotalFailedRequestsToFuncInWindow(f 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)) - val, warn, err := promApiClient.client.Query(context.Background(), queryString, time.Now()) + val, warn, err := promApiClient.client.Query(ctx, queryString, time.Now()) if err != nil { return 0, errors.Wrapf(err, "error querying prometheus") } diff --git a/pkg/controller/functionApi.go b/pkg/controller/functionApi.go index bb97d3b5..dafd3d71 100644 --- a/pkg/controller/functionApi.go +++ b/pkg/controller/functionApi.go @@ -332,21 +332,21 @@ func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) { } // 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 { a.respondWithError(w, errors.Wrapf(err, "error getting container logs")) 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) for _, container := range pod.Spec.Containers { podLogOpts := apiv1.PodLogOptions{Container: container.Name} // Only the env container, not fetcher 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 { return errors.Wrapf(err, "error streaming pod log") }