Skip to content

Commit

Permalink
Use tempaltes for profile analyzer
Browse files Browse the repository at this point in the history
  • Loading branch information
Shambugouda annigeri committed Aug 3, 2022
1 parent b827a0d commit 8c2ebf7
Show file tree
Hide file tree
Showing 3 changed files with 116 additions and 69 deletions.
1 change: 1 addition & 0 deletions api/v1beta1/zz_generated.deepcopy.go
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions api/v1beta2/zz_generated.deepcopy.go
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

183 changes: 114 additions & 69 deletions kubectl-fdb/cmd/profile_analyzer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,29 +21,54 @@
package cmd

import (
"context"
"github.com/FoundationDB/fdb-kubernetes-operator/internal"
"github.com/google/uuid"
"bytes"
ctx "context"
"github.com/spf13/cobra"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
"io"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer/yaml"
"k8s.io/cli-runtime/pkg/genericclioptions"
"k8s.io/utils/pointer"
"log"
"sigs.k8s.io/controller-runtime/pkg/client"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/restmapper"
"strconv"
"text/template"

yamlutil "k8s.io/apimachinery/pkg/util/yaml"
"log"
)

type profileConfig struct {
Namespace string
ClusterName string
JobName string
CommandArgs string
}

func newProfileAnalyzerCmd(streams genericclioptions.IOStreams) *cobra.Command {
o := newFDBOptions(streams)

cmd := &cobra.Command{
Use: "analyze-profile",
Use: "analyze-profile",
Short: "Analyze FDB shards to find the busiest team",
Long: "Analyze FDB shards to find the busiest team",
Long: "Analyze FDB shards to find the busiest team",
RunE: func(cmd *cobra.Command, args []string) error {
config, err := o.configFlags.ToRESTConfig()
if err != nil {
return err
}
clientSet, err := kubernetes.NewForConfig(config)
if err != nil {
return err
}
dynamicConfig, err := dynamic.NewForConfig(config)
if err != nil {
return err
}
clusterName, err := cmd.Flags().GetString("fdb-cluster")
if err != nil {
return err
Expand All @@ -60,20 +85,21 @@ func newProfileAnalyzerCmd(streams genericclioptions.IOStreams) *cobra.Command {
if err != nil {
return err
}
kubeClient, err := getKubeClient(o)
templateName, err := cmd.Flags().GetString("template-name")
if err != nil {
return err
}

namespace, err := getNamespace(*o.configFlags.Namespace)
if err != nil {
return err
}

return runProfileAnalyzer(kubeClient, namespace, clusterName, startTime, endTime, topRequests)
return runProfileAnalyzer(clientSet, dynamicConfig, namespace, clusterName, startTime, endTime, topRequests, templateName)
},
Example: `
# Run the profiler for cluster-1. We require --cluster option explicitly because analyze commands take lot many arguments.
kubectl fdb analyze-profile -c cluster-1 --star-time "01:01 20/07/2022 BST" --end-time "01:30 20/07/2022 BST" --top-requests 100
kubectl fdb analyze-profile -c cluster-1 --start-time "01:01 20/07/2022 BST" --end-time "01:30 20/07/2022 BST" --top-requests 100
`,
}
cmd.SetOut(o.Out)
Expand All @@ -83,6 +109,7 @@ kubectl fdb analyze-profile -c cluster-1 --star-time "01:01 20/07/2022 BST" --en
cmd.Flags().StringP("fdb-cluster", "c", "", "cluster name for running hot shard tool.")
cmd.Flags().String("start-time", "", "start time for the analyzing transaction '01:30 30/07/2022 BST'")
cmd.Flags().String("end-time", "", "end time for analyzing the transaction '02:30 30/07/2022 BST'")
cmd.Flags().String("template-name", "", "Name of the Job template")
cmd.Flags().Int("top-requests", 100, "")
err := cmd.MarkFlagRequired("fdb-cluster")
if err != nil {
Expand All @@ -93,61 +120,79 @@ kubectl fdb analyze-profile -c cluster-1 --star-time "01:01 20/07/2022 BST" --en
return cmd
}

func runProfileAnalyzer(kubeClient client.Client, namespace string, clusterName string, startTime string, endTime string, topRequests int) error {
job := &batchv1.Job{
ObjectMeta: metav1.ObjectMeta{
Name: clusterName + "-hot-shard-" + uuid.New().String(),
Namespace: namespace,
},
Spec: batchv1.JobSpec{
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
SecurityContext: &corev1.PodSecurityContext{
RunAsUser: pointer.Int64(4059),
RunAsGroup: pointer.Int64(4059),
FSGroup: pointer.Int64(4059),
},
func runProfileAnalyzer(kubeClient *kubernetes.Clientset, dynamicConfig dynamic.Interface, namespace string, clusterName string, startTime string, endTime string, topRequests int, templateName string) error {
pc := profileConfig{
Namespace: namespace,
ClusterName: clusterName,
JobName: clusterName + "-hot-shard-tool",
CommandArgs: " -C " + " /var/dynamic-conf/fdb.cluster" + " -s \"" + startTime + "\"" + " -e \"" + endTime + "\"" + " --filter-get-range " + " --top-requests " + strconv.Itoa(topRequests),
}
t, err := template.ParseFiles(templateName)
if err != nil {
return err
}
buf := bytes.Buffer{}
err = t.Execute(&buf, pc)
if err != nil {
return err
}
decoder := yamlutil.NewYAMLOrJSONDecoder(&buf, 100000)

Containers: []corev1.Container{
{
Name: "profile-analyzer",
Image: "fdb-profile-analyzer",
Command: []string{"python3", "./transaction_profiling_analyzer.py"},
Args: []string{"-C", "$FDB_CLUSTER_FILE", "--start-time", startTime, "--end-time", endTime,
"--filter-get-range", "--top-requests", strconv.Itoa(topRequests),
},
Env: []corev1.EnvVar{
{Name: "FDB_CLUSTER_FILE", Value: "/var/dynamic-conf/fdb.cluster"},
},
Resources: corev1.ResourceRequirements{
Requests: corev1.ResourceList{
"cpu": resource.MustParse("1"),
"memory": resource.MustParse("1Gi"),
},
},
VolumeMounts: []corev1.VolumeMount{
{Name: "config-map", MountPath: "/var/dynamic-conf"},
},
},
},
RestartPolicy: corev1.RestartPolicyNever,
Volumes: []corev1.Volume{{
Name: "config-map",
VolumeSource: corev1.VolumeSource{ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{Name: clusterName + "-config"},
Items: []corev1.KeyToPath{
{Key: internal.ClusterFileKey, Path: "fdb.cluster"},
},
}},
},
{
Name: "dynamic-conf", VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}},
},
},
},
TTLSecondsAfterFinished: pointer.Int32(7200),
},
for {
var rawObj runtime.RawExtension
err := decoder.Decode(&rawObj)
if err != nil {
if err == io.EOF {
break
}
return err
}
obj, gvk, err := yaml.NewDecodingSerializer(unstructured.UnstructuredJSONScheme).
Decode(rawObj.Raw, nil, nil)
if err != nil {
log.Println("NewDecodingSerializer Error ->" + err.Error())
return err
}
unstructuredMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
if err != nil {
log.Println("ToUnstructured Error ->" + err.Error())
return err
}
unstructuredObj := &unstructured.Unstructured{Object: unstructuredMap}
gr, err := restmapper.GetAPIGroupResources(kubeClient.Discovery())
if err != nil {
return err
}
mapper := restmapper.NewDiscoveryRESTMapper(gr)
mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
if err != nil {
log.Println("rest mapping Error ->" + err.Error())
return err
}
var dynamicInterface dynamic.ResourceInterface
if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
if unstructuredObj.GetNamespace() == "" {
unstructuredObj.SetNamespace(namespace)
}
dynamicInterface = dynamicConfig.Resource(mapping.Resource).
Namespace(unstructuredObj.GetNamespace())
} else {
dynamicInterface = dynamicConfig.Resource(mapping.Resource)
}
if _, err := dynamicInterface.Create(
ctx.Background(),
unstructuredObj,
metav1.CreateOptions{
FieldManager: "fdb-hot-shard-tool",
}); err != nil {
if k8serrors.IsAlreadyExists(err) {
log.Printf("%s job already present, Delete the job and re-run.", pc.JobName)
continue
}
log.Println("DynamicInterface Error ->" + err.Error())
return err
}
}
log.Printf("Creating job %s", job.Name)
return kubeClient.Create(context.TODO(), job)
log.Printf("%s Job created.", pc.JobName)
return nil
}

0 comments on commit 8c2ebf7

Please sign in to comment.