feat(executor): Log verb kind statusCode for executor Kubernetes API requests

cmd/argoexec/commands/root.go

@@ -14,6 +14,7 @@ import (
 	"github.com/argoproj/argo/v3"
 	"github.com/argoproj/argo/v3/util"
 	"github.com/argoproj/argo/v3/util/cmd"
+	"github.com/argoproj/argo/v3/util/logs"
 	"github.com/argoproj/argo/v3/workflow/common"
 	"github.com/argoproj/argo/v3/workflow/executor"
 	"github.com/argoproj/argo/v3/workflow/executor/docker"
@@ -74,6 +75,8 @@ func initExecutor() *executor.WorkflowExecutor {
 	config, err := clientConfig.ClientConfig()
 	checkErr(err)
 
+	logs.AddK8SLogTransportWrapper(config) // lets log all request as we should typically do < 5 per pod, so this is will show up problems
+
 	namespace, _, err := clientConfig.Namespace()
 	checkErr(err)
 

util/k8s/parse.go

@@ -0,0 +1,38 @@
+package k8s
+
+import (
+	"net/http"
+	"strings"
+)
+
+func ParseRequest(r *http.Request) (verb string, kind string) {
+	i := strings.Index(r.URL.Path, "/v") + 1
+	path := strings.Split(r.URL.Path[i:], "/")
+	n := len(path)
+
+	verb = map[string]string{
+		http.MethodGet:    "List",
+		http.MethodPost:   "Create",
+		http.MethodDelete: "Delete",
+		http.MethodPatch:  "Patch",
+		http.MethodPut:    "Update",
+	}[r.Method]
+
+	if r.URL.Query().Get("watch") != "" {
+		verb = "Watch"
+	} else if verb == "List" && n%2 == 1 {
+		verb = "Get"
+	} else if verb == "Delete" && n%2 == 0 {
+		verb = "DeleteCollection"
+	}
+
+	kind = "Unknown"
+	switch verb {
+	case "List", "Watch", "Create", "DeleteCollection":
+		kind = path[n-1]
+	case "Get", "Delete", "Patch", "Update":
+		kind = path[n-2]
+	}
+
+	return verb, kind
+}

workflow/metrics/k8s_request_total_metric_test.go → util/k8s/parse_test.go

@@ -1,4 +1,4 @@
-package metrics
+package k8s
 
 import (
 	"net/http"
@@ -8,7 +8,7 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func Test_parseRequest(t *testing.T) {
+func Test_ParseRequest(t *testing.T) {
 	for _, tt := range []struct {
 		name     string
 		method   string
@@ -26,7 +26,7 @@ func Test_parseRequest(t *testing.T) {
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			x, _ := url.Parse(tt.url)
-			verb, kind := parseRequest(&http.Request{Method: tt.method, URL: x})
+			verb, kind := ParseRequest(&http.Request{Method: tt.method, URL: x})
 			assert.Equal(t, tt.wantVerb, verb)
 			assert.Equal(t, tt.wantKind, kind)
 		})

util/logs/log-k8s-requests.go

@@ -0,0 +1,34 @@
+package logs
+
+import (
+	"net/http"
+
+	log "github.com/sirupsen/logrus"
+	"k8s.io/client-go/rest"
+
+	"github.com/argoproj/argo/v3/util/k8s"
+)
+
+type k8sLogRoundTripper struct {
+	roundTripper http.RoundTripper
+}
+
+func (m k8sLogRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
+	x, err := m.roundTripper.RoundTrip(r)
+	if x != nil {
+		verb, kind := k8s.ParseRequest(r)
+		log.Infof("%s %s %d", verb, kind, x.StatusCode)
+	}
+	return x, err
+}
+
+func AddK8SLogTransportWrapper(config *rest.Config) *rest.Config {
+	wrap := config.WrapTransport
+	config.WrapTransport = func(rt http.RoundTripper) http.RoundTripper {
+		if wrap != nil {
+			rt = wrap(rt)
+		}
+		return &k8sLogRoundTripper{roundTripper: rt}
+	}
+	return config
+}

workflow/metrics/k8s_request_total_metric.go

@@ -3,10 +3,11 @@ package metrics
 import (
 	"net/http"
 	"strconv"
-	"strings"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"k8s.io/client-go/rest"
+
+	"github.com/argoproj/argo/v3/util/k8s"
 )
 
 var (
@@ -28,44 +29,12 @@ type metricsRoundTripper struct {
 func (m metricsRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
 	x, err := m.roundTripper.RoundTrip(r)
 	if x != nil {
-		verb, kind := parseRequest(r)
+		verb, kind := k8s.ParseRequest(r)
 		K8sRequestTotalMetric.WithLabelValues(kind, verb, strconv.Itoa(x.StatusCode)).Inc()
 	}
 	return x, err
 }
 
-func parseRequest(r *http.Request) (verb string, kind string) {
-	i := strings.Index(r.URL.Path, "/v") + 1
-	path := strings.Split(r.URL.Path[i:], "/")
-	n := len(path)
-
-	verb = map[string]string{
-		http.MethodGet:    "List",
-		http.MethodPost:   "Create",
-		http.MethodDelete: "Delete",
-		http.MethodPatch:  "Patch",
-		http.MethodPut:    "Update",
-	}[r.Method]
-
-	if r.URL.Query().Get("watch") != "" {
-		verb = "Watch"
-	} else if verb == "List" && n%2 == 1 {
-		verb = "Get"
-	} else if verb == "Delete" && n%2 == 0 {
-		verb = "DeleteCollection"
-	}
-
-	kind = "Unknown"
-	switch verb {
-	case "List", "Watch", "Create", "DeleteCollection":
-		kind = path[n-1]
-	case "Get", "Delete", "Patch", "Update":
-		kind = path[n-2]
-	}
-
-	return verb, kind
-}
-
 func AddMetricsTransportWrapper(config *rest.Config) *rest.Config {
 	wrap := config.WrapTransport
 	config.WrapTransport = func(rt http.RoundTripper) http.RoundTripper {

workflow/util/util.go

@@ -359,7 +359,7 @@ func ResumeWorkflow(ctx context.Context, wfIf v1alpha1.WorkflowInterface, hydrat
 
 			err = hydrator.Hydrate(wf)
 			if err != nil {
-				return false, err
+				return true, err
 			}
 
 			workflowUpdated := false