From b5ce3276b92fe4d711a0e13ca448abde0db5f722 Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Wed, 25 Dec 2024 15:26:04 +0530 Subject: [PATCH 1/3] fix: removed workload watchers --- common/ip_resolver.go | 624 +++++++++--------------------------------- 1 file changed, 131 insertions(+), 493 deletions(-) diff --git a/common/ip_resolver.go b/common/ip_resolver.go index 662f252a..d4ce31ab 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -10,20 +10,21 @@ import ( "time" "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" "k8s.io/klog/v2" lrucache "github.com/hashicorp/golang-lru/v2" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" - appsv1 "k8s.io/api/apps/v1" - batchv1 "k8s.io/api/batch/v1" - "k8s.io/api/batch/v1beta1" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -const MAX_RESOLVED_DNS = 10000 // arbitrary limit +const MAX_RESOLVED_DNS = 10000 +const MAX_PODS = 20000 + var reregisterWatchSleepDuration = 1 * time.Second var ( @@ -38,13 +39,6 @@ var ( type clusterSnapshot struct { Pods sync.Map // map[types.UID]v1.Pod Nodes sync.Map // map[types.UID]v1.Node - ReplicaSets sync.Map // map[types.UID]appsv1.ReplicaSet - DaemonSets sync.Map // map[types.UID]appsv1.DaemonSet - StatefulSets sync.Map // map[types.UID]appsv1.StatefulSet - Jobs sync.Map // map[types.UID]batchv1.Job - Services sync.Map // map[types.UID]v1.Service - Deployments sync.Map // map[types.UID]appsv1.Deployment - CronJobs sync.Map // map[types.UID]batchv1.CronJob or batchv1beta.CronJob PodDescriptors sync.Map // map[types.UID]Workload } @@ -55,7 +49,7 @@ type K8sIPResolver struct { stopSignal chan bool shouldResolveDns bool dnsResolvedIps *lrucache.Cache[string, string] - podIpsMap sync.Map + podIpsMap *lrucache.Cache[string, Workload] instanceMetaMap sync.Map } @@ -85,6 +79,11 @@ func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPRe } else { dnsCache = nil } + podsCache, err := lrucache.New[string, Workload](MAX_RESOLVED_DNS) + if err != nil { + return nil, err + } + return &K8sIPResolver{ clientset: clientset, snapshot: clusterSnapshot{}, @@ -92,17 +91,14 @@ func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPRe stopSignal: make(chan bool), shouldResolveDns: resolveDns, dnsResolvedIps: dnsCache, - podIpsMap: sync.Map{}, + podIpsMap: podsCache, instanceMetaMap: sync.Map{}, }, nil } func (resolver *K8sIPResolver) ResolveActualIP(ip string) Workload { - if val, ok := resolver.podIpsMap.Load(ip); ok { - entry, ok := val.(Workload) - if ok { - return entry - } + if val, ok := resolver.podIpsMap.Get(ip); ok { + return val } host := ip @@ -137,11 +133,8 @@ func (resolver *K8sIPResolver) ResolveIP(ip string) Workload { log.Printf("type confusion in ipsMap") } - if val, ok := resolver.podIpsMap.Load(ip); ok { - entry, ok := val.(Workload) - if ok { - return entry - } + if val, ok := resolver.podIpsMap.Get(ip); ok { + return val } host := ip @@ -269,97 +262,6 @@ func (resolver *K8sIPResolver) StartWatching() error { watchEventsCounter.WithLabelValues("node").Inc() resolver.handleNodeWatchEvent(&nodeEvent) } - case replicasetsEvent, ok := <-replicasetsWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("replicaset").Inc() - replicasetsWatcher, err = resolver.clientset.AppsV1().ReplicaSets("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("replicaset").Inc() - resolver.handleReplicaSetWatchEvent(&replicasetsEvent) - } - case daemonsetsEvent, ok := <-daemonsetsWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("daemonset").Inc() - daemonsetsWatcher, err = resolver.clientset.AppsV1().DaemonSets("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("daemonset").Inc() - resolver.handleDaemonSetWatchEvent(&daemonsetsEvent) - } - case statefulsetsEvent, ok := <-statefulsetsWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("statefulset").Inc() - statefulsetsWatcher, err = resolver.clientset.AppsV1().StatefulSets("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("statefulset").Inc() - resolver.handleStatefulSetWatchEvent(&statefulsetsEvent) - } - case jobsEvent, ok := <-jobsWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("job").Inc() - jobsWatcher, err = resolver.clientset.BatchV1().Jobs("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("job").Inc() - resolver.handleJobsWatchEvent(&jobsEvent) - } - case servicesEvent, ok := <-servicesWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("service").Inc() - servicesWatcher, err = resolver.clientset.CoreV1().Services("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("service").Inc() - resolver.handleServicesWatchEvent(&servicesEvent) - } - case deploymentsEvent, ok := <-deploymentsWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("deployment").Inc() - deploymentsWatcher, err = resolver.clientset.AppsV1().Deployments("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("deployment").Inc() - resolver.handleDeploymentsWatchEvent(&deploymentsEvent) - } - case cronjobsEvent, ok := <-cronJobsWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("cronjob").Inc() - cronJobsWatcher, err = resolver.startCronjobWatcher() - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("cronjob").Inc() - resolver.handleCronJobsWatchEvent(&cronjobsEvent) - } } } }() @@ -374,6 +276,64 @@ func (resolver *K8sIPResolver) StartWatching() error { return nil } +func (resolver *K8sIPResolver) StartWatchingV2(ip string, entry *Workload) { + + factory := informers.NewSharedInformerFactory(resolver.clientset, time.Minute*10) + // Define individual informers + podInformer := factory.Core().V1().Pods().Informer() + nodeInformer := factory.Core().V1().Nodes().Informer() + + resolver.addPodHandlers(podInformer) + resolver.addNodeHandlers(nodeInformer) +} +func (resolver *K8sIPResolver) addPodHandlers(podInformer cache.SharedIndexInformer) { + podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + pod := obj.(*v1.Pod) + log.Printf("Pod added: %s/%s", pod.Namespace, pod.Name) + resolver.handlePodAdd(pod) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + newPod := newObj.(*v1.Pod) + log.Printf("Pod updated: %s/%s", newPod.Namespace, newPod.Name) + resolver.handlePodAdd(newPod) + }, + DeleteFunc: func(obj interface{}) { + pod := obj.(*v1.Pod) + log.Printf("Pod deleted: %s/%s", pod.Namespace, pod.Name) + resolver.snapshot.Pods.Delete(pod.UID) + resolver.snapshot.PodDescriptors.Delete(pod.UID) + for _, podIp := range pod.Status.PodIPs { + resolver.podIpsMap.Remove(podIp.IP) + } + }, + }) +} + +func (resolver *K8sIPResolver) addNodeHandlers(nodeInformer cache.SharedIndexInformer) { + nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + node := obj.(*v1.Node) + shouldReturn := resolver.handleNodeEvent(node) + if shouldReturn { + return + } + }, + UpdateFunc: func(oldObj, newObj interface{}) { + node := newObj.(*v1.Node) + shouldReturn := resolver.handleNodeEvent(node) + if shouldReturn { + return + } + }, + DeleteFunc: func(obj interface{}) { + node := obj.(*v1.Node) + resolver.snapshot.Nodes.Delete(node.UID) + resolver.instanceMetaMap.Delete(node.Name) + }, + }) +} + func (resolver *K8sIPResolver) startCronjobWatcher() (watch.Interface, error) { cronJobsWatcher, err := resolver.clientset.BatchV1().CronJobs("").Watch(context.Background(), metav1.ListOptions{}) if err != nil { @@ -389,76 +349,12 @@ func (resolver *K8sIPResolver) StopWatching() { func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { switch podEvent.Type { - case watch.Added: - pod, ok := podEvent.Object.(*v1.Pod) - if !ok { - return - } - resolver.snapshot.Pods.Store(pod.UID, *pod) - entry := resolver.resolvePodDescriptor(pod) - instanceMeta, ok := resolver.instanceMetaMap.Load(pod.Spec.NodeName) - region := "" - zone := "" - if ok { - instanceMeta, ok := instanceMeta.(InstanceMeta) - if ok { - region = instanceMeta.Region - zone = instanceMeta.Zone - } else { - log.Printf("Type confusion in instance meta for node %s", pod.Spec.NodeName) - } - } else { - log.Printf("Missing instance meta for node %s", pod.Spec.NodeName) - resolver.instanceMetaMap.Range(func(key, value interface{}) bool { - log.Printf("key: %s, value: %v", key, value) - return true - }) - } - for _, podIp := range pod.Status.PodIPs { - resolver.storeWorkloadsIP(podIp.IP, &entry) - podWorkload := Workload{ - Name: pod.Name, - Namespace: pod.Namespace, - Kind: "pod", - Region: region, - Zone: zone, - Instance: pod.Spec.NodeName, - } - resolver.storePodsIP(podIp.IP, &podWorkload) - } - case watch.Modified: + case watch.Added, watch.Modified: pod, ok := podEvent.Object.(*v1.Pod) if !ok { return } - resolver.snapshot.Pods.Store(pod.UID, *pod) - entry := resolver.resolvePodDescriptor(pod) - instanceMeta, ok := resolver.instanceMetaMap.Load(pod.Spec.NodeName) - region := "" - zone := "" - if ok { - instanceMeta, ok := instanceMeta.(InstanceMeta) - if ok { - region = instanceMeta.Region - zone = instanceMeta.Zone - } else { - log.Printf("Type confusion in instance meta for node %s", pod.Spec.NodeName) - } - } else { - log.Printf("Missing instance meta for node %s", pod.Spec.NodeName) - } - for _, podIp := range pod.Status.PodIPs { - resolver.storeWorkloadsIP(podIp.IP, &entry) - podWorkload := Workload{ - Name: pod.Name, - Namespace: pod.Namespace, - Kind: "pod", - Region: region, - Zone: zone, - Instance: pod.Spec.NodeName, - } - resolver.storePodsIP(podIp.IP, &podWorkload) - } + resolver.handlePodAdd(pod) case watch.Deleted: if val, ok := podEvent.Object.(*v1.Pod); ok { resolver.snapshot.Pods.Delete(val.UID) @@ -467,159 +363,77 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { } } -func (resolver *K8sIPResolver) handleNodeWatchEvent(nodeEvent *watch.Event) { - switch nodeEvent.Type { - case watch.Added, watch.Modified: - node, ok := nodeEvent.Object.(*v1.Node) - if !ok { - return - } - resolver.snapshot.Nodes.Store(node.UID, *node) - // extract region and zone from attributes - region := node.Labels["topology.kubernetes.io/region"] - zone := node.Labels["topology.kubernetes.io/zone"] - meta := InstanceMeta{ - Region: region, - Zone: zone, - Instance: node.Name, - } - resolver.instanceMetaMap.Store(node.Name, meta) - for _, nodeAddress := range node.Status.Addresses { - resolver.storeWorkloadsIP(nodeAddress.Address, &Workload{ - Name: node.Name, - Namespace: "node", - Kind: "node", - Region: region, - Zone: zone, - Instance: node.Name, - }) - } - case watch.Deleted: - if val, ok := nodeEvent.Object.(*v1.Node); ok { - resolver.snapshot.Nodes.Delete(val.UID) - resolver.instanceMetaMap.Delete(val.Name) - } - } -} - -func (resolver *K8sIPResolver) handleReplicaSetWatchEvent(replicasetsEvent *watch.Event) { - switch replicasetsEvent.Type { - case watch.Added: - if val, ok := replicasetsEvent.Object.(*appsv1.ReplicaSet); ok { - resolver.snapshot.ReplicaSets.Store(val.UID, *val) - } - case watch.Deleted: - if val, ok := replicasetsEvent.Object.(*appsv1.ReplicaSet); ok { - resolver.snapshot.ReplicaSets.Delete(val.UID) - } - } -} - -func (resolver *K8sIPResolver) handleDaemonSetWatchEvent(daemonsetsEvent *watch.Event) { - switch daemonsetsEvent.Type { - case watch.Added: - if val, ok := daemonsetsEvent.Object.(*appsv1.DaemonSet); ok { - resolver.snapshot.DaemonSets.Store(val.UID, *val) - } - case watch.Deleted: - if val, ok := daemonsetsEvent.Object.(*appsv1.DaemonSet); ok { - resolver.snapshot.DaemonSets.Delete(val.UID) - } - } -} - -func (resolver *K8sIPResolver) handleStatefulSetWatchEvent(statefulsetsEvent *watch.Event) { - switch statefulsetsEvent.Type { - case watch.Added: - if val, ok := statefulsetsEvent.Object.(*appsv1.StatefulSet); ok { - resolver.snapshot.StatefulSets.Store(val.UID, *val) - } - case watch.Deleted: - if val, ok := statefulsetsEvent.Object.(*appsv1.StatefulSet); ok { - resolver.snapshot.StatefulSets.Delete(val.UID) +func (resolver *K8sIPResolver) handlePodAdd(pod *v1.Pod) bool { + resolver.snapshot.Pods.Store(pod.UID, *pod) + entry := resolver.resolvePodDescriptor(pod) + instanceMeta, ok := resolver.instanceMetaMap.Load(pod.Spec.NodeName) + region := "" + zone := "" + if ok { + instanceMeta, ok := instanceMeta.(InstanceMeta) + if ok { + region = instanceMeta.Region + zone = instanceMeta.Zone + } else { + log.Printf("Type confusion in instance meta for node %s", pod.Spec.NodeName) } } -} - -func (resolver *K8sIPResolver) handleJobsWatchEvent(jobsEvent *watch.Event) { - switch jobsEvent.Type { - case watch.Added: - if val, ok := jobsEvent.Object.(*batchv1.Job); ok { - resolver.snapshot.Jobs.Store(val.UID, *val) - } - case watch.Deleted: - if val, ok := jobsEvent.Object.(*batchv1.Job); ok { - resolver.snapshot.Jobs.Delete(val.UID) + for _, podIp := range pod.Status.PodIPs { + resolver.storeWorkloadsIP(podIp.IP, &entry) + podWorkload := Workload{ + Name: pod.Name, + Namespace: pod.Namespace, + Kind: "pod", + Region: region, + Zone: zone, + Instance: pod.Spec.NodeName, } + resolver.storePodsIP(podIp.IP, &podWorkload) } + return false } -func (resolver *K8sIPResolver) handleServicesWatchEvent(servicesEvent *watch.Event) { - switch servicesEvent.Type { +func (resolver *K8sIPResolver) handleNodeWatchEvent(nodeEvent *watch.Event) { + switch nodeEvent.Type { case watch.Added, watch.Modified: - service, ok := servicesEvent.Object.(*v1.Service) + // extract region and zone from attributes + node, ok := nodeEvent.Object.(*v1.Node) if !ok { return } - resolver.snapshot.Services.Store(service.UID, *service) - - // services has (potentially multiple) ClusterIP - workload := Workload{ - Name: service.Name, - Namespace: service.Namespace, - Kind: "Service", - Zone: "", - Region: "", - Instance: "", - } - - // TODO maybe try to match service to workload - for _, clusterIp := range service.Spec.ClusterIPs { - if clusterIp != "None" { - _, ok := resolver.ipsMap.Load(clusterIp) - if !ok { - resolver.storeWorkloadsIP(clusterIp, &workload) - } - } - } - case watch.Deleted: - if val, ok := servicesEvent.Object.(*v1.Service); ok { - resolver.snapshot.Services.Delete(val.UID) - } - } -} - -func (resolver *K8sIPResolver) handleDeploymentsWatchEvent(deploymentsEvent *watch.Event) { - switch deploymentsEvent.Type { - case watch.Added: - if val, ok := deploymentsEvent.Object.(*appsv1.Deployment); ok { - resolver.snapshot.Deployments.Store(val.UID, *val) + shouldReturn := resolver.handleNodeEvent(node) + if shouldReturn { + return } case watch.Deleted: - if val, ok := deploymentsEvent.Object.(*appsv1.Deployment); ok { - resolver.snapshot.Deployments.Delete(val.UID) + if val, ok := nodeEvent.Object.(*v1.Node); ok { + resolver.snapshot.Nodes.Delete(val.UID) + resolver.instanceMetaMap.Delete(val.Name) } } } -func (resolver *K8sIPResolver) handleCronJobsWatchEvent(cronjobsEvent *watch.Event) { - switch cronjobsEvent.Type { - case watch.Added: - if val, ok := cronjobsEvent.Object.(*batchv1.CronJob); ok { - resolver.snapshot.CronJobs.Store(val.UID, *val) - } - if val, ok := cronjobsEvent.Object.(*v1beta1.CronJob); ok { - resolver.snapshot.CronJobs.Store(val.UID, *val) - } - - case watch.Deleted: - if val, ok := cronjobsEvent.Object.(*batchv1.CronJob); ok { - resolver.snapshot.CronJobs.Delete(val.UID) - } - if val, ok := cronjobsEvent.Object.(*v1beta1.CronJob); ok { - resolver.snapshot.CronJobs.Delete(val.UID) - } +func (resolver *K8sIPResolver) handleNodeEvent(node *v1.Node) bool { + resolver.snapshot.Nodes.Store(node.UID, *node) + region := node.Labels["topology.kubernetes.io/region"] + zone := node.Labels["topology.kubernetes.io/zone"] + meta := InstanceMeta{ + Region: region, + Zone: zone, + Instance: node.Name, + } + resolver.instanceMetaMap.Store(node.Name, meta) + for _, nodeAddress := range node.Status.Addresses { + resolver.storeWorkloadsIP(nodeAddress.Address, &Workload{ + Name: node.Name, + Namespace: "node", + Kind: "node", + Region: region, + Zone: zone, + Instance: node.Name, + }) } + return false } func (resolver *K8sIPResolver) getResolvedClusterSnapshot() error { @@ -660,104 +474,12 @@ func (resolver *K8sIPResolver) getFullClusterSnapshot() error { resolver.instanceMetaMap.Store(node.Name, meta) resolver.snapshot.Nodes.Store(node.UID, node) } - - replicasets, err := resolver.clientset.AppsV1().ReplicaSets("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting replicasets, aborting snapshot update") - } - for _, rs := range replicasets.Items { - resolver.snapshot.ReplicaSets.Store(rs.ObjectMeta.UID, rs) - } - - daemonsets, err := resolver.clientset.AppsV1().DaemonSets("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting daemonsets, aborting snapshot update") - } - for _, ds := range daemonsets.Items { - resolver.snapshot.DaemonSets.Store(ds.ObjectMeta.UID, ds) - } - - statefulsets, err := resolver.clientset.AppsV1().StatefulSets("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting statefulsets, aborting snapshot update") - } - for _, ss := range statefulsets.Items { - resolver.snapshot.StatefulSets.Store(ss.ObjectMeta.UID, ss) - } - - jobs, err := resolver.clientset.BatchV1().Jobs("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting jobs, aborting snapshot update") - } - for _, job := range jobs.Items { - resolver.snapshot.Jobs.Store(job.ObjectMeta.UID, job) - } - - services, err := resolver.clientset.CoreV1().Services("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting services, aborting snapshot update") - } - for _, service := range services.Items { - resolver.snapshot.Services.Store(service.UID, service) - } - - deployments, err := resolver.clientset.AppsV1().Deployments("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting deployments, aborting snapshot update") - } - for _, deployment := range deployments.Items { - resolver.snapshot.Deployments.Store(deployment.UID, deployment) - } - - cronJobs, err := resolver.clientset.BatchV1().CronJobs("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - cronJobs, err := resolver.clientset.BatchV1beta1().CronJobs("").List(context.Background(), metav1.ListOptions{}) - if err != nil { - return errors.New("error getting cronjobs, aborting snapshot update") - } - for _, cronJob := range cronJobs.Items { - resolver.snapshot.CronJobs.Store(cronJob.UID, cronJob) - } - } - for _, cronJob := range cronJobs.Items { - resolver.snapshot.CronJobs.Store(cronJob.UID, cronJob) - } - return nil } // add mapping from ip to resolved host to an existing map, // based on the given cluster snapshot func (resolver *K8sIPResolver) updateIpMapping() { - // because IP collisions may occur and lead to overwrites in the map, the order is important - // we go from less "favorable" to more "favorable" - - // services -> running pods -> nodes - - resolver.snapshot.Services.Range(func(key any, val any) bool { - service, ok := val.(v1.Service) - if !ok { - log.Printf("Type confusion in services map") - return true // continue - } - // services has (potentially multiple) ClusterIP - workload := Workload{ - Name: service.Name, - Namespace: service.Namespace, - Kind: "Service", - Zone: "", - Region: "", - Instance: "", - } - - // TODO maybe try to match service to workload - for _, clusterIp := range service.Spec.ClusterIPs { - if clusterIp != "None" { - resolver.storeWorkloadsIP(clusterIp, &workload) - } - } - return true - }) - resolver.snapshot.Nodes.Range(func(key any, value any) bool { node, ok := value.(v1.Node) if !ok { @@ -840,80 +562,7 @@ func (resolver *K8sIPResolver) storeWorkloadsIP(ip string, newWorkload *Workload } func (resolver *K8sIPResolver) storePodsIP(ip string, newWorkload *Workload) { - resolver.podIpsMap.Store(ip, *newWorkload) -} - -// an ugly function to go up one level in hierarchy. maybe there's a better way to do it -// the snapshot is maintained to avoid using an API request for each resolving -func (resolver *K8sIPResolver) getControllerOfOwner(originalOwner *metav1.OwnerReference) (*metav1.OwnerReference, error) { - switch originalOwner.Kind { - case "ReplicaSet": - replicaSetVal, ok := resolver.snapshot.ReplicaSets.Load(originalOwner.UID) - if !ok { - return nil, errors.New("Missing replicaset for UID " + string(originalOwner.UID)) - } - replicaSet, ok := replicaSetVal.(appsv1.ReplicaSet) - if !ok { - return nil, errors.New("type confusion in replicasets map") - } - return metav1.GetControllerOf(&replicaSet), nil - case "DaemonSet": - daemonSetVal, ok := resolver.snapshot.DaemonSets.Load(originalOwner.UID) - if !ok { - return nil, errors.New("Missing daemonset for UID " + string(originalOwner.UID)) - } - daemonSet, ok := daemonSetVal.(appsv1.DaemonSet) - if !ok { - return nil, errors.New("type confusion in daemonsets map") - } - return metav1.GetControllerOf(&daemonSet), nil - case "StatefulSet": - statefulSetVal, ok := resolver.snapshot.StatefulSets.Load(originalOwner.UID) - if !ok { - return nil, errors.New("Missing statefulset for UID " + string(originalOwner.UID)) - } - statefulSet, ok := statefulSetVal.(appsv1.StatefulSet) - if !ok { - return nil, errors.New("type confusion in statefulsets map") - } - return metav1.GetControllerOf(&statefulSet), nil - case "Job": - jobVal, ok := resolver.snapshot.Jobs.Load(originalOwner.UID) - if !ok { - return nil, errors.New("Missing job for UID " + string(originalOwner.UID)) - } - job, ok := jobVal.(batchv1.Job) - if !ok { - return nil, errors.New("type confusion in jobs map") - } - return metav1.GetControllerOf(&job), nil - case "Deployment": - deploymentVal, ok := resolver.snapshot.Deployments.Load(originalOwner.UID) - if !ok { - return nil, errors.New("Missing deployment for UID " + string(originalOwner.UID)) - } - deployment, ok := deploymentVal.(appsv1.Deployment) - if !ok { - return nil, errors.New("type confusion in deployments map") - } - return metav1.GetControllerOf(&deployment), nil - case "CronJob": - cronJobVal, ok := resolver.snapshot.CronJobs.Load(originalOwner.UID) - if !ok { - return nil, errors.New("Missing cronjob for UID " + string(originalOwner.UID)) - } - cronJob, ok := cronJobVal.(batchv1.CronJob) - if !ok { - cronJob, ok := cronJobVal.(v1beta1.CronJob) - if !ok { - return nil, errors.New("type confusion in cronjobs map") - } - return metav1.GetControllerOf(&cronJob), nil - } - - return metav1.GetControllerOf(&cronJob), nil - } - return nil, errors.New("Unsupported kind for lookup - " + originalOwner.Kind) + resolver.podIpsMap.Add(ip, *newWorkload) } func (resolver *K8sIPResolver) resolvePodDescriptor(pod *v1.Pod) Workload { @@ -928,17 +577,6 @@ func (resolver *K8sIPResolver) resolvePodDescriptor(pod *v1.Pod) Workload { name := pod.Name namespace := pod.Namespace kind := "pod" - owner := metav1.GetControllerOf(pod) - // climbing up the owners' hierarchy. if an error occurs, we take the data we got and save - // the error to know we shouldn't save this resolution to the descriptors map and retry later. - for owner != nil { - name = owner.Name - kind = owner.Kind - owner, err = resolver.getControllerOfOwner(owner) - if err != nil { - log.Printf("Warning: couldn't retrieve owner of %v - %v. This might happen when starting up", name, err) - } - } instanceMeta, ok := resolver.instanceMetaMap.Load(pod.Spec.NodeName) region := "" zone := "" From 27fe0e267d7d8a188040b2c6138876336914369f Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Fri, 27 Dec 2024 11:38:31 +0530 Subject: [PATCH 2/3] fix: fix for invalid utf char error --- common/ip_resolver.go | 545 ++++++++++++++++++++++++++-------------- containers/container.go | 9 +- containers/l7.go | 2 +- tracing/tracing.go | 15 +- 4 files changed, 373 insertions(+), 198 deletions(-) diff --git a/common/ip_resolver.go b/common/ip_resolver.go index d4ce31ab..e6ee9c33 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -9,36 +9,31 @@ import ( "sync" "time" - "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" - "k8s.io/klog/v2" lrucache "github.com/hashicorp/golang-lru/v2" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promauto" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + "k8s.io/api/batch/v1beta1" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -const MAX_RESOLVED_DNS = 10000 -const MAX_PODS = 20000 - -var reregisterWatchSleepDuration = 1 * time.Second - -var ( - watchEventsCounter = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "nudgbee_watcher_events_count", - }, []string{"object_type"}) - watchResetsCounter = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "nudgebee_watcher_resets_count", - }, []string{"object_type"}) -) +const MAX_RESOLVED_DNS = 10000 // arbitrary limit +const MAX_PODS_LIMIT = 30000 // arbitrary limit type clusterSnapshot struct { Pods sync.Map // map[types.UID]v1.Pod Nodes sync.Map // map[types.UID]v1.Node + ReplicaSets sync.Map // map[types.UID]appsv1.ReplicaSet + DaemonSets sync.Map // map[types.UID]appsv1.DaemonSet + StatefulSets sync.Map // map[types.UID]appsv1.StatefulSet + Jobs sync.Map // map[types.UID]batchv1.Job + Services sync.Map // map[types.UID]v1.Service + Deployments sync.Map // map[types.UID]appsv1.Deployment + CronJobs sync.Map // map[types.UID]batchv1.CronJob or batchv1beta.CronJob PodDescriptors sync.Map // map[types.UID]Workload } @@ -46,7 +41,7 @@ type K8sIPResolver struct { clientset kubernetes.Interface snapshot clusterSnapshot ipsMap sync.Map - stopSignal chan bool + stopSignal chan struct{} shouldResolveDns bool dnsResolvedIps *lrucache.Cache[string, string] podIpsMap *lrucache.Cache[string, Workload] @@ -79,7 +74,8 @@ func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPRe } else { dnsCache = nil } - podsCache, err := lrucache.New[string, Workload](MAX_RESOLVED_DNS) + var podCache *lrucache.Cache[string, Workload] + podCache, err := lrucache.New[string, Workload](MAX_PODS_LIMIT) if err != nil { return nil, err } @@ -88,10 +84,10 @@ func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPRe clientset: clientset, snapshot: clusterSnapshot{}, ipsMap: sync.Map{}, - stopSignal: make(chan bool), + stopSignal: make(chan struct{}), shouldResolveDns: resolveDns, dnsResolvedIps: dnsCache, - podIpsMap: podsCache, + podIpsMap: podCache, instanceMetaMap: sync.Map{}, }, nil } @@ -173,119 +169,155 @@ func (resolver *K8sIPResolver) CacheDNS(ip string, dns string) Workload { } func (resolver *K8sIPResolver) StartWatching() error { - // register watchers - podsWatcher, err := resolver.clientset.CoreV1().Pods("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching pods changes - %v", err) - } - - nodesWatcher, err := resolver.clientset.CoreV1().Nodes().Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching nodes changes - %v", err) - } - - replicasetsWatcher, err := resolver.clientset.AppsV1().ReplicaSets("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching replicasets changes - %v", err) - } - - daemonsetsWatcher, err := resolver.clientset.AppsV1().DaemonSets("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching daemonsets changes - %v", err) - } - - statefulsetsWatcher, err := resolver.clientset.AppsV1().StatefulSets("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching statefulsets changes - %v", err) - } - - jobsWatcher, err := resolver.clientset.BatchV1().Jobs("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching jobs changes - %v", err) - } - - servicesWatcher, err := resolver.clientset.CoreV1().Services("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching services changes - %v", err) - } - - deploymentsWatcher, err := resolver.clientset.AppsV1().Deployments("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return fmt.Errorf("error watching deployments changes - %v", err) - } + factory := informers.NewSharedInformerFactory(resolver.clientset, time.Minute*10) + // Define individual informers + podInformer := factory.Core().V1().Pods().Informer() + nodeInformer := factory.Core().V1().Nodes().Informer() + replicaSetInformer := factory.Apps().V1().ReplicaSets().Informer() + daemonSetInformer := factory.Apps().V1().DaemonSets().Informer() + statefulSetInformer := factory.Apps().V1().StatefulSets().Informer() + jobInformer := factory.Batch().V1().Jobs().Informer() + serviceInformer := factory.Core().V1().Services().Informer() + deploymentInformer := factory.Apps().V1().Deployments().Informer() - cronJobsWatcher, err := resolver.startCronjobWatcher() - if err != nil { - klog.Errorf("failed to init watcher %v", err) - return fmt.Errorf("error watching cronjobs changes - %v", err) - } - log.Printf("registered IP watcher") - // invoke a watching function - go func() { - for { - select { - case <-resolver.stopSignal: - podsWatcher.Stop() - nodesWatcher.Stop() - replicasetsWatcher.Stop() - daemonsetsWatcher.Stop() - statefulsetsWatcher.Stop() - jobsWatcher.Stop() - servicesWatcher.Stop() - deploymentsWatcher.Stop() - cronJobsWatcher.Stop() - return - case podEvent, ok := <-podsWatcher.ResultChan(): - { - - if !ok { - watchResetsCounter.WithLabelValues("pod").Inc() - podsWatcher, err = resolver.clientset.CoreV1().Pods("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("pod").Inc() - resolver.handlePodWatchEvent(&podEvent) - } - case nodeEvent, ok := <-nodesWatcher.ResultChan(): - { - if !ok { - watchResetsCounter.WithLabelValues("node").Inc() - nodesWatcher, err = resolver.clientset.CoreV1().Nodes().Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - time.Sleep(reregisterWatchSleepDuration) - } - continue - } - watchEventsCounter.WithLabelValues("node").Inc() - resolver.handleNodeWatchEvent(&nodeEvent) - } - } - } - }() + resolver.addPodHandlers(podInformer) + resolver.addNodeHandlers(nodeInformer) + resolver.addReplicaSetHandlers(replicaSetInformer) + resolver.addDaemonSetHandlers(daemonSetInformer) + resolver.addStatefulSetHandlers(statefulSetInformer) + resolver.addJobHandlers(jobInformer) + resolver.addServiceHandlers(serviceInformer) + resolver.addDeploymentHandlers(deploymentInformer) // get initial state - err = resolver.getResolvedClusterSnapshot() + err := resolver.getResolvedClusterSnapshot() if err != nil { - resolver.StopWatching() return fmt.Errorf("error retrieving cluster's initial state: %v", err) } - + factory.Start(resolver.stopSignal) // runs in background return nil } -func (resolver *K8sIPResolver) StartWatchingV2(ip string, entry *Workload) { +func (resolve *K8sIPResolver) addReplicaSetHandlers(replicaSetInformer cache.SharedIndexInformer) { + replicaSetInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + replicaSet := obj.(*appsv1.ReplicaSet) + log.Printf("ReplicaSet added: %s/%s", replicaSet.Namespace, replicaSet.Name) + resolve.snapshot.ReplicaSets.Store(replicaSet.UID, *replicaSet) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + replicaSet := newObj.(*appsv1.ReplicaSet) + log.Printf("ReplicaSet updated: %s/%s", replicaSet.Namespace, replicaSet.Name) + resolve.snapshot.ReplicaSets.Store(replicaSet.UID, *replicaSet) + }, + DeleteFunc: func(obj interface{}) { + replicaSet := obj.(*appsv1.ReplicaSet) + log.Printf("ReplicaSet deleted: %s/%s", replicaSet.Namespace, replicaSet.Name) + resolve.snapshot.ReplicaSets.Delete(replicaSet.UID) + }, + }) +} - factory := informers.NewSharedInformerFactory(resolver.clientset, time.Minute*10) - // Define individual informers - podInformer := factory.Core().V1().Pods().Informer() - nodeInformer := factory.Core().V1().Nodes().Informer() +func (resolve *K8sIPResolver) addDaemonSetHandlers(daemonSetInformer cache.SharedIndexInformer) { + daemonSetInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + daemonSet := obj.(*appsv1.DaemonSet) + log.Printf("DaemonSet added: %s/%s", daemonSet.Namespace, daemonSet.Name) + resolve.snapshot.DaemonSets.Store(daemonSet.UID, *daemonSet) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + daemonSet := newObj.(*appsv1.DaemonSet) + log.Printf("DaemonSet updated: %s/%s", daemonSet.Namespace, daemonSet.Name) + resolve.snapshot.DaemonSets.Store(daemonSet.UID, *daemonSet) + }, + DeleteFunc: func(obj interface{}) { + daemonSet := obj.(*appsv1.DaemonSet) + log.Printf("DaemonSet deleted: %s/%s", daemonSet.Namespace, daemonSet.Name) + resolve.snapshot.DaemonSets.Delete(daemonSet.UID) + }, + }) +} - resolver.addPodHandlers(podInformer) - resolver.addNodeHandlers(nodeInformer) +func (resolve *K8sIPResolver) addStatefulSetHandlers(statefulSetInformer cache.SharedIndexInformer) { + statefulSetInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + statefulSet := obj.(*appsv1.StatefulSet) + log.Printf("StatefulSet added: %s/%s", statefulSet.Namespace, statefulSet.Name) + resolve.snapshot.StatefulSets.Store(statefulSet.UID, *statefulSet) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + statefulSet := newObj.(*appsv1.StatefulSet) + log.Printf("StatefulSet updated: %s/%s", statefulSet.Namespace, statefulSet.Name) + resolve.snapshot.StatefulSets.Store(statefulSet.UID, *statefulSet) + }, + DeleteFunc: func(obj interface{}) { + statefulSet := obj.(*appsv1.StatefulSet) + log.Printf("StatefulSet deleted: %s/%s", statefulSet.Namespace, statefulSet.Name) + resolve.snapshot.StatefulSets.Delete(statefulSet.UID) + }, + }) +} + +func (resolve *K8sIPResolver) addJobHandlers(jobInformer cache.SharedIndexInformer) { + jobInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + job := obj.(*batchv1.Job) + log.Printf("Job added: %s/%s", job.Namespace, job.Name) + resolve.snapshot.Jobs.Store(job.UID, *job) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + job := newObj.(*batchv1.Job) + log.Printf("Job updated: %s/%s", job.Namespace, job.Name) + resolve.snapshot.Jobs.Store(job.UID, *job) + }, + DeleteFunc: func(obj interface{}) { + job := obj.(*batchv1.Job) + log.Printf("Job deleted: %s/%s", job.Namespace, job.Name) + resolve.snapshot.Jobs.Delete(job.UID) + }, + }) } + +func (resolve *K8sIPResolver) addServiceHandlers(serviceInformer cache.SharedIndexInformer) { + serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + service := obj.(*v1.Service) + log.Printf("Service added: %s/%s", service.Namespace, service.Name) + resolve.snapshot.Services.Store(service.UID, *service) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + service := newObj.(*v1.Service) + log.Printf("Service updated: %s/%s", service.Namespace, service.Name) + resolve.snapshot.Services.Store(service.UID, *service) + }, + DeleteFunc: func(obj interface{}) { + service := obj.(*v1.Service) + log.Printf("Service deleted: %s/%s", service.Namespace, service.Name) + resolve.snapshot.Services.Delete(service.UID) + }, + }) +} + +func (resolve *K8sIPResolver) addDeploymentHandlers(deploymentInformer cache.SharedIndexInformer) { + deploymentInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + deployment := obj.(*appsv1.Deployment) + log.Printf("Deployment added: %s/%s", deployment.Namespace, deployment.Name) + resolve.snapshot.Deployments.Store(deployment.UID, *deployment) + }, + UpdateFunc: func(oldObj, newObj interface{}) { + deployment := newObj.(*appsv1.Deployment) + log.Printf("Deployment updated: %s/%s", deployment.Namespace, deployment.Name) + resolve.snapshot.Deployments.Store(deployment.UID, *deployment) + }, + DeleteFunc: func(obj interface{}) { + deployment := obj.(*appsv1.Deployment) + log.Printf("Deployment deleted: %s/%s", deployment.Namespace, deployment.Name) + resolve.snapshot.Deployments.Delete(deployment.UID) + }, + }) +} + func (resolver *K8sIPResolver) addPodHandlers(podInformer cache.SharedIndexInformer) { podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { @@ -310,59 +342,6 @@ func (resolver *K8sIPResolver) addPodHandlers(podInformer cache.SharedIndexInfor }) } -func (resolver *K8sIPResolver) addNodeHandlers(nodeInformer cache.SharedIndexInformer) { - nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - node := obj.(*v1.Node) - shouldReturn := resolver.handleNodeEvent(node) - if shouldReturn { - return - } - }, - UpdateFunc: func(oldObj, newObj interface{}) { - node := newObj.(*v1.Node) - shouldReturn := resolver.handleNodeEvent(node) - if shouldReturn { - return - } - }, - DeleteFunc: func(obj interface{}) { - node := obj.(*v1.Node) - resolver.snapshot.Nodes.Delete(node.UID) - resolver.instanceMetaMap.Delete(node.Name) - }, - }) -} - -func (resolver *K8sIPResolver) startCronjobWatcher() (watch.Interface, error) { - cronJobsWatcher, err := resolver.clientset.BatchV1().CronJobs("").Watch(context.Background(), metav1.ListOptions{}) - if err != nil { - return resolver.clientset.BatchV1beta1().CronJobs("").Watch(context.Background(), metav1.ListOptions{}) - } - - return cronJobsWatcher, nil -} - -func (resolver *K8sIPResolver) StopWatching() { - resolver.stopSignal <- true -} - -func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { - switch podEvent.Type { - case watch.Added, watch.Modified: - pod, ok := podEvent.Object.(*v1.Pod) - if !ok { - return - } - resolver.handlePodAdd(pod) - case watch.Deleted: - if val, ok := podEvent.Object.(*v1.Pod); ok { - resolver.snapshot.Pods.Delete(val.UID) - resolver.snapshot.PodDescriptors.Delete(val.UID) - } - } -} - func (resolver *K8sIPResolver) handlePodAdd(pod *v1.Pod) bool { resolver.snapshot.Pods.Store(pod.UID, *pod) entry := resolver.resolvePodDescriptor(pod) @@ -393,24 +372,28 @@ func (resolver *K8sIPResolver) handlePodAdd(pod *v1.Pod) bool { return false } -func (resolver *K8sIPResolver) handleNodeWatchEvent(nodeEvent *watch.Event) { - switch nodeEvent.Type { - case watch.Added, watch.Modified: - // extract region and zone from attributes - node, ok := nodeEvent.Object.(*v1.Node) - if !ok { - return - } - shouldReturn := resolver.handleNodeEvent(node) - if shouldReturn { - return - } - case watch.Deleted: - if val, ok := nodeEvent.Object.(*v1.Node); ok { - resolver.snapshot.Nodes.Delete(val.UID) - resolver.instanceMetaMap.Delete(val.Name) - } - } +func (resolver *K8sIPResolver) addNodeHandlers(nodeInformer cache.SharedIndexInformer) { + nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + node := obj.(*v1.Node) + shouldReturn := resolver.handleNodeEvent(node) + if shouldReturn { + return + } + }, + UpdateFunc: func(oldObj, newObj interface{}) { + node := newObj.(*v1.Node) + shouldReturn := resolver.handleNodeEvent(node) + if shouldReturn { + return + } + }, + DeleteFunc: func(obj interface{}) { + node := obj.(*v1.Node) + resolver.snapshot.Nodes.Delete(node.UID) + resolver.instanceMetaMap.Delete(node.Name) + }, + }) } func (resolver *K8sIPResolver) handleNodeEvent(node *v1.Node) bool { @@ -474,12 +457,104 @@ func (resolver *K8sIPResolver) getFullClusterSnapshot() error { resolver.instanceMetaMap.Store(node.Name, meta) resolver.snapshot.Nodes.Store(node.UID, node) } + + replicasets, err := resolver.clientset.AppsV1().ReplicaSets("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting replicasets, aborting snapshot update") + } + for _, rs := range replicasets.Items { + resolver.snapshot.ReplicaSets.Store(rs.ObjectMeta.UID, rs) + } + + daemonsets, err := resolver.clientset.AppsV1().DaemonSets("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting daemonsets, aborting snapshot update") + } + for _, ds := range daemonsets.Items { + resolver.snapshot.DaemonSets.Store(ds.ObjectMeta.UID, ds) + } + + statefulsets, err := resolver.clientset.AppsV1().StatefulSets("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting statefulsets, aborting snapshot update") + } + for _, ss := range statefulsets.Items { + resolver.snapshot.StatefulSets.Store(ss.ObjectMeta.UID, ss) + } + + jobs, err := resolver.clientset.BatchV1().Jobs("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting jobs, aborting snapshot update") + } + for _, job := range jobs.Items { + resolver.snapshot.Jobs.Store(job.ObjectMeta.UID, job) + } + + services, err := resolver.clientset.CoreV1().Services("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting services, aborting snapshot update") + } + for _, service := range services.Items { + resolver.snapshot.Services.Store(service.UID, service) + } + + deployments, err := resolver.clientset.AppsV1().Deployments("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting deployments, aborting snapshot update") + } + for _, deployment := range deployments.Items { + resolver.snapshot.Deployments.Store(deployment.UID, deployment) + } + + cronJobs, err := resolver.clientset.BatchV1().CronJobs("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + cronJobs, err := resolver.clientset.BatchV1beta1().CronJobs("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting cronjobs, aborting snapshot update") + } + for _, cronJob := range cronJobs.Items { + resolver.snapshot.CronJobs.Store(cronJob.UID, cronJob) + } + } + for _, cronJob := range cronJobs.Items { + resolver.snapshot.CronJobs.Store(cronJob.UID, cronJob) + } + return nil } // add mapping from ip to resolved host to an existing map, // based on the given cluster snapshot func (resolver *K8sIPResolver) updateIpMapping() { + // because IP collisions may occur and lead to overwrites in the map, the order is important + // we go from less "favorable" to more "favorable" - + // services -> running pods -> nodes + + resolver.snapshot.Services.Range(func(key any, val any) bool { + service, ok := val.(v1.Service) + if !ok { + log.Printf("Type confusion in services map") + return true // continue + } + // services has (potentially multiple) ClusterIP + workload := Workload{ + Name: service.Name, + Namespace: service.Namespace, + Kind: "Service", + Zone: "", + Region: "", + Instance: "", + } + + // TODO maybe try to match service to workload + for _, clusterIp := range service.Spec.ClusterIPs { + if clusterIp != "None" { + resolver.storeWorkloadsIP(clusterIp, &workload) + } + } + return true + }) + resolver.snapshot.Nodes.Range(func(key any, value any) bool { node, ok := value.(v1.Node) if !ok { @@ -565,6 +640,79 @@ func (resolver *K8sIPResolver) storePodsIP(ip string, newWorkload *Workload) { resolver.podIpsMap.Add(ip, *newWorkload) } +// an ugly function to go up one level in hierarchy. maybe there's a better way to do it +// the snapshot is maintained to avoid using an API request for each resolving +func (resolver *K8sIPResolver) getControllerOfOwner(originalOwner *metav1.OwnerReference) (*metav1.OwnerReference, error) { + switch originalOwner.Kind { + case "ReplicaSet": + replicaSetVal, ok := resolver.snapshot.ReplicaSets.Load(originalOwner.UID) + if !ok { + return nil, errors.New("Missing replicaset for UID " + string(originalOwner.UID)) + } + replicaSet, ok := replicaSetVal.(appsv1.ReplicaSet) + if !ok { + return nil, errors.New("type confusion in replicasets map") + } + return metav1.GetControllerOf(&replicaSet), nil + case "DaemonSet": + daemonSetVal, ok := resolver.snapshot.DaemonSets.Load(originalOwner.UID) + if !ok { + return nil, errors.New("Missing daemonset for UID " + string(originalOwner.UID)) + } + daemonSet, ok := daemonSetVal.(appsv1.DaemonSet) + if !ok { + return nil, errors.New("type confusion in daemonsets map") + } + return metav1.GetControllerOf(&daemonSet), nil + case "StatefulSet": + statefulSetVal, ok := resolver.snapshot.StatefulSets.Load(originalOwner.UID) + if !ok { + return nil, errors.New("Missing statefulset for UID " + string(originalOwner.UID)) + } + statefulSet, ok := statefulSetVal.(appsv1.StatefulSet) + if !ok { + return nil, errors.New("type confusion in statefulsets map") + } + return metav1.GetControllerOf(&statefulSet), nil + case "Job": + jobVal, ok := resolver.snapshot.Jobs.Load(originalOwner.UID) + if !ok { + return nil, errors.New("Missing job for UID " + string(originalOwner.UID)) + } + job, ok := jobVal.(batchv1.Job) + if !ok { + return nil, errors.New("type confusion in jobs map") + } + return metav1.GetControllerOf(&job), nil + case "Deployment": + deploymentVal, ok := resolver.snapshot.Deployments.Load(originalOwner.UID) + if !ok { + return nil, errors.New("Missing deployment for UID " + string(originalOwner.UID)) + } + deployment, ok := deploymentVal.(appsv1.Deployment) + if !ok { + return nil, errors.New("type confusion in deployments map") + } + return metav1.GetControllerOf(&deployment), nil + case "CronJob": + cronJobVal, ok := resolver.snapshot.CronJobs.Load(originalOwner.UID) + if !ok { + return nil, errors.New("Missing cronjob for UID " + string(originalOwner.UID)) + } + cronJob, ok := cronJobVal.(batchv1.CronJob) + if !ok { + cronJob, ok := cronJobVal.(v1beta1.CronJob) + if !ok { + return nil, errors.New("type confusion in cronjobs map") + } + return metav1.GetControllerOf(&cronJob), nil + } + + return metav1.GetControllerOf(&cronJob), nil + } + return nil, errors.New("Unsupported kind for lookup - " + originalOwner.Kind) +} + func (resolver *K8sIPResolver) resolvePodDescriptor(pod *v1.Pod) Workload { existing, ok := resolver.snapshot.PodDescriptors.Load(pod.UID) if ok { @@ -577,6 +725,17 @@ func (resolver *K8sIPResolver) resolvePodDescriptor(pod *v1.Pod) Workload { name := pod.Name namespace := pod.Namespace kind := "pod" + owner := metav1.GetControllerOf(pod) + // climbing up the owners' hierarchy. if an error occurs, we take the data we got and save + // the error to know we shouldn't save this resolution to the descriptors map and retry later. + for owner != nil { + name = owner.Name + kind = owner.Kind + owner, err = resolver.getControllerOfOwner(owner) + if err != nil { + log.Printf("Warning: couldn't retrieve owner of %v - %v. This might happen when starting up", name, err) + } + } instanceMeta, ok := resolver.instanceMetaMap.Load(pod.Spec.NodeName) region := "" zone := "" diff --git a/containers/container.go b/containers/container.go index eeb046c8..e38cf22a 100644 --- a/containers/container.go +++ b/containers/container.go @@ -7,6 +7,7 @@ import ( "strings" "sync" "time" + "unicode/utf8" "encoding/base64" @@ -792,7 +793,13 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R headers = base64.StdEncoding.EncodeToString([]byte(headersStr)) } method = req.Method - uri = req.URL.Path + if utf8.ValidString(req.URL.Path) { + uri = req.URL.Path + } else { + // remove non-utf8 characters + klog.Warningf("Non-utf8 characters in uri %q", req.URL.Path) + uri = string([]rune(req.URL.Path)) + } host = req.Host } if dns, ok := iqfqdn[conn.Dest.IP()]; ok { diff --git a/containers/l7.go b/containers/l7.go index b5407a76..ca5109d0 100644 --- a/containers/l7.go +++ b/containers/l7.go @@ -82,7 +82,7 @@ func (s L7Stats) get(protocol l7.Protocol, destination, actualDestination netadd if host != "" { constLabels["destination_workload_name"] = host } else { - log.Printf("Failed to parse host %s , %q", host, err) + log.Printf("Failed to parse host %s , %v", string(r.Payload), err) } } hOpts := L7Latency[protocol] diff --git a/tracing/tracing.go b/tracing/tracing.go index d1874c7e..2e18ce86 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -104,7 +104,6 @@ func NewTrace(containerId string, destination netaddr.IPPort, srcWorkload common zone := "" node := "" if actualDstWorkload.Zone != "" { - zone = actualDstWorkload.Zone } if actualDstWorkload.Region != "" { @@ -167,8 +166,13 @@ func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time if utf8.ValidString(path) { requestPath = path } + requestHost := "" + if utf8.ValidString(host) { + requestHost = host + } + t.createSpan(method, duration, status >= 400, - semconv.HTTPURL(fmt.Sprintf("http://%s%s", host, requestPath)), + semconv.HTTPURL(fmt.Sprintf("http://%s%s", requestHost, requestPath)), semconv.HTTPMethod(method), semconv.HTTPStatusCode(int(status)), semconv.HTTPRequestContentLength(int(requestSize)), @@ -192,11 +196,16 @@ func (t *Trace) Http2Request(method, path, scheme string, status l7.Status, dura if scheme == "" { scheme = "unknown" } + requestPayload := "" + if utf8.ValidString(payload) { + requestPayload = payload + } + t.createSpan(method, duration, status > 400, semconv.HTTPURL(fmt.Sprintf("%s://%s%s", scheme, t.destination.String(), path)), semconv.HTTPMethod(method), semconv.HTTPStatusCode(int(status)), - attribute.Key("http.request_payload").String(payload), + attribute.Key("http.request_payload").String(requestPayload), ) } From 7adab04cb86af0c984e4dcc2d2b06caa1b269a2a Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Fri, 27 Dec 2024 12:56:20 +0530 Subject: [PATCH 3/3] fix: fix for signal handle --- common/ip_resolver.go | 4 ++++ main.go | 16 +++++++++++++++- 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/common/ip_resolver.go b/common/ip_resolver.go index e6ee9c33..87d6e9d0 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -168,6 +168,10 @@ func (resolver *K8sIPResolver) CacheDNS(ip string, dns string) Workload { } } +func (resolver *K8sIPResolver) StopWatching() { + close(resolver.stopSignal) +} + func (resolver *K8sIPResolver) StartWatching() error { factory := informers.NewSharedInformerFactory(resolver.clientset, time.Minute*10) // Define individual informers diff --git a/main.go b/main.go index 9ea2da0a..64117a70 100644 --- a/main.go +++ b/main.go @@ -6,8 +6,10 @@ import ( "net/http" _ "net/http/pprof" "os" + "os/signal" "runtime" "strings" + "syscall" "github.com/coroot/coroot-node-agent/common" "github.com/coroot/coroot-node-agent/containers" @@ -137,12 +139,24 @@ func main() { } err = resolver.StartWatching() if err != nil { - log.Fatalf("Error watching cluster's state: %v", err) + klog.Errorf("Error starting resolver: %v", err) } + go func() { + sigChannel := make(chan os.Signal, 1) + defer close(sigChannel) + + signal.Notify(sigChannel, os.Interrupt, syscall.SIGTERM) + <-sigChannel + + klog.Infoln("Received signal, shutting down") + resolver.StopWatching() + os.Exit(0) // Ensure graceful termination + }() hostname, kv, err := uname() if err != nil { klog.Exitln("failed to get uname:", err) } + klog.Infoln("hostname:", hostname) klog.Infoln("kernel version:", kv)