feat(kprofefe): bootstrap daemon

Fixed #7

This PR contains a first iteration for kprofefe the daemon that runs
inside a kubernetes cluster as a cronjob and keeps collecting and
persisting profiles in profefe.

Signed-off-by: Gianluca Arbezzano <[email protected]>

Author: Gianluca Arbezzano

.goreleaser.yml

@@ -8,6 +8,13 @@ builds:
 - id: "kubectl-profefe"
   main: ./cmd/kubectl-profefe
   binary: kubectl-profefe
+  env:
+    - CGO_ENABLED=0
+- id: "kprofefe"
+  main: ./cmd/kprofefe
+  binary: kprofefe
+  env:
+    - CGO_ENABLED=0
 archives:
 - replacements:
     darwin: Darwin
@@ -25,3 +32,14 @@ changelog:
     exclude:
     - '^docs:'
     - '^test:'
+# .goreleaser.yml
+dockers:
+  -
+    binaries:
+    - kprofefe
+    goos: linux
+    goarch: amd64
+    image_templates:
+    - "profefe/kprofefe:latest"
+    - "profefe/kprofefe:v{{ .Tag }}"
+    dockerfile: Dockerfile.gorelease

Dockerfile.gorelease

@@ -0,0 +1,5 @@
+FROM alpine
+
+COPY kprofefe /
+
+ENTRYPOINT ["/kprofefe"]

cmd/kprofefe/main.go

@@ -0,0 +1,19 @@
+package main
+
+import (
+	"os"
+
+	"github.com/gianarb/kube-profefe/pkg/cmd"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	_ "k8s.io/client-go/plugin/pkg/client/auth/oidc"
+)
+
+func main() {
+	rootCmd := cmd.NewKProfefeCmd(genericclioptions.IOStreams{
+		In:     os.Stdin,
+		Out:    os.Stdout,
+		ErrOut: os.Stderr,
+	})
+
+	rootCmd.Execute()
+}

pkg/cmd/capture.go

@@ -168,7 +168,7 @@ profefe.`)
 
 func writeProfiles(ctx context.Context, pClient *profefe.Client, profiles map[pprofutil.Profile]*profile.Profile, target corev1.Pod) error {
 	for profileType, profile := range profiles {
-		profefeType := profefe.NewProfileTypeFromString(profile.PeriodType.Type)
+		profefeType := profefe.NewProfileTypeFromString(profileType.String())
 		if profefeType == profefe.UnknownProfile {
 			println("unknown profile type: :" + profile.PeriodType.Type)
 			continue

pkg/cmd/kprofefe.go

@@ -0,0 +1,131 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+
+	"github.com/gianarb/kube-profefe/pkg/kubeutil"
+	"github.com/gianarb/kube-profefe/pkg/pprofutil"
+	"github.com/gianarb/kube-profefe/pkg/profefe"
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+func NewKProfefeCmd(streams genericclioptions.IOStreams) *cobra.Command {
+	flags := pflag.NewFlagSet("kubectl-profefe", pflag.ExitOnError)
+	pflag.CommandLine = flags
+
+	kubeConfigFlags := genericclioptions.NewConfigFlags(false)
+	kubeResouceBuilderFlags := genericclioptions.NewResourceBuilderFlags()
+
+	cmd := &cobra.Command{
+		Use:   "kprofefe",
+		Short: "kprofefe collects profiles from inside a kubernetes cluster",
+		PersistentPreRun: func(c *cobra.Command, args []string) {
+			c.SetOutput(streams.ErrOut)
+		},
+		Run: func(cmd *cobra.Command, args []string) {
+			var config *rest.Config
+			var err error
+
+			config, err = rest.InClusterConfig()
+			if err != nil {
+				config, err = kubeConfigFlags.ToRESTConfig()
+			}
+			if err != nil {
+				panic(err)
+			}
+			if config == nil {
+				panic("woww")
+			}
+			// creates the clientset
+			clientset, err := kubernetes.NewForConfig(config)
+			if err != nil {
+				panic(err.Error())
+			}
+
+			// Contains the pool of pods that we need to gather profiles from
+			selectedPods := []corev1.Pod{}
+
+			namespace := kubeutil.GetNamespaceFromKubernetesFlags(kubeConfigFlags, kubeResouceBuilderFlags)
+
+			// If the arguments are more than zero we should check by pod name
+			// (args == resourceName)
+			if len(args) > 0 {
+				for _, podName := range args {
+					pod, err := kubeutil.GetPodByName(clientset, namespace, podName, metav1.GetOptions{})
+					if err != nil {
+						println(err.Error())
+						continue
+					}
+					selectedPods = append(selectedPods, *pod)
+				}
+			} else {
+				selectedPods, err = kubeutil.GetSelectedPods(clientset, namespace, metav1.ListOptions{
+					LabelSelector: *kubeResouceBuilderFlags.LabelSelector,
+				})
+				if err != nil {
+					println(err.Error())
+					os.Exit(1)
+				}
+			}
+
+			// If the selectedPods are zero there is nothing to do.
+			if len(selectedPods) == 0 {
+				println("there are not pod that matches your research")
+				os.Exit(1)
+			}
+
+			println(fmt.Sprintf("selected %d pod/s", len(selectedPods)))
+
+			pClient := profefe.NewClient(profefe.Config{
+				HostPort: ProfefeHostPort,
+			}, http.Client{})
+
+			for _, target := range selectedPods {
+				targetPort := pprofutil.GetProfefePortByPod(target)
+				profiles, err := pprofutil.GatherAllByPod(context.Background(), fmt.Sprintf("http://%s", target.Status.PodIP), target, targetPort)
+				if err != nil {
+					panic(err)
+				}
+				for profileType, profile := range profiles {
+					profefeType := profefe.NewProfileTypeFromString(profileType.String())
+					if profefeType == profefe.UnknownProfile {
+						println("unknown profile type: :" + profileType.String())
+						continue
+					}
+					saved, err := pClient.SavePprof(context.Background(), profefe.SavePprofRequest{
+						Profile:    profile,
+						Service:    target.Name,
+						InstanceID: target.Status.HostIP,
+						Type:       profefeType,
+						Labels: map[string]string{
+							"namespace": target.Namespace,
+							"from":      "kube-profefe",
+						},
+					})
+					if err != nil {
+						println(fmt.Sprintf("%s type=%s profile_type=%s", err.Error(), profefeType, profile.PeriodType.Type))
+					} else {
+						println(fmt.Sprintf("%s/api/0/profiles/%s type=%s", ProfefeHostPort, saved.Body.ID, profefeType))
+					}
+				}
+			}
+		},
+	}
+
+	flags.AddFlagSet(cmd.PersistentFlags())
+	flags.StringVar(&ProfefeHostPort, "profefe-hostport", "http://localhost:10100", `where profefe is located`)
+	kubeConfigFlags.AddFlags(flags)
+	kubeResouceBuilderFlags.WithLabelSelector("")
+	kubeResouceBuilderFlags.AddFlags(flags)
+
+	return cmd
+}

pkg/pprofutil/pprof.go

@@ -3,6 +3,7 @@ package pprofutil
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	corev1 "k8s.io/api/core/v1"
 
@@ -11,14 +12,15 @@ import (
 
 const (
 	DefaultProfilePath = "/debug/pprof"
+	DefaultProfilePort = 6060
 )
 
 func GatherAllByPod(ctx context.Context, host string, pod corev1.Pod, forwardedPort int) (map[Profile]*profile.Profile, error) {
 	path := DefaultProfilePath
 	if rawPath, ok := pod.Annotations["profefe.com/path"]; ok && rawPath != "" {
 		path = rawPath
 	}
-	return GatherAll(ctx, fmt.Sprintf("%s:%d/%s", host, forwardedPort, path))
+	return GatherAll(ctx, fmt.Sprintf("%s:%d%s", host, forwardedPort, path))
 }
 
 // GatherAll downloads all profile types from address.
@@ -51,3 +53,13 @@ func GatherAll(ctx context.Context, addr string) (map[Profile]*profile.Profile,
 
 	return profs, err
 }
+
+func GetProfefePortByPod(pod corev1.Pod) int {
+	port := DefaultProfilePort
+	if rawPort, ok := pod.Annotations["profefe.com/port"]; ok && rawPort != "" {
+		if i, err := strconv.Atoi(rawPort); err == nil {
+			port = i
+		}
+	}
+	return port
+}