diff --git a/common/ip_resolver.go b/common/ip_resolver.go new file mode 100644 index 00000000..c8fe4717 --- /dev/null +++ b/common/ip_resolver.go @@ -0,0 +1,830 @@ +package common + +import ( + "context" + "errors" + "fmt" + "log" + "net" + "sync" + "time" + + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes" + "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 +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"}) +) + +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 +} + +type K8sIPResolver struct { + clientset kubernetes.Interface + snapshot clusterSnapshot + ipsMap sync.Map + stopSignal chan bool + shouldResolveDns bool + dnsResolvedIps *lrucache.Cache[string, string] + podIpsMap sync.Map +} + +type Workload struct { + Name string + Namespace string + Kind string +} + +func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPResolver, error) { + var dnsCache *lrucache.Cache[string, string] + if resolveDns { + var err error + dnsCache, err = lrucache.New[string, string](MAX_RESOLVED_DNS) + if err != nil { + return nil, err + } + } else { + dnsCache = nil + } + return &K8sIPResolver{ + clientset: clientset, + snapshot: clusterSnapshot{}, + ipsMap: sync.Map{}, + stopSignal: make(chan bool), + shouldResolveDns: resolveDns, + dnsResolvedIps: dnsCache, + podIpsMap: sync.Map{}, + }, nil +} + +// resolve the given IP from the resolver's cache +// if not available, return the IP itself. +func (resolver *K8sIPResolver) ResolveActualIP(ip string) Workload { + if val, ok := resolver.podIpsMap.Load(ip); ok { + entry, ok := val.(Workload) + if ok { + return entry + } + } + host := ip + + if resolver.shouldResolveDns { + val, ok := resolver.dnsResolvedIps.Get(ip) + if ok { + host = val + } else { + hosts, err := net.LookupAddr(ip) + if err == nil && len(hosts) > 0 { + host = hosts[0] + } + resolver.dnsResolvedIps.Add(ip, host) + } + } + return Workload{ + Name: host, + Namespace: "external", + Kind: "external", + } +} + +func (resolver *K8sIPResolver) ResolveIP(ip string) Workload { + if val, ok := resolver.ipsMap.Load(ip); ok { + entry, ok := val.(Workload) + if ok { + return entry + } + log.Printf("type confusion in ipsMap") + } + + if val, ok := resolver.podIpsMap.Load(ip); ok { + entry, ok := val.(Workload) + if ok { + return entry + } + } + host := ip + + if resolver.shouldResolveDns { + val, ok := resolver.dnsResolvedIps.Get(ip) + if ok { + host = val + } else { + hosts, err := net.LookupAddr(ip) + if err == nil && len(hosts) > 0 { + host = hosts[0] + } + resolver.dnsResolvedIps.Add(ip, host) + } + } + return Workload{ + Name: host, + Namespace: "external", + Kind: "external", + } +} + +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) + } + + 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) + } + 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) + } + } + } + }() + + // get initial state + err = resolver.getResolvedClusterSnapshot() + if err != nil { + resolver.StopWatching() + return fmt.Errorf("error retrieving cluster's initial state: %v", err) + } + + return nil +} + +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: + pod, ok := podEvent.Object.(*v1.Pod) + if !ok { + return + } + resolver.snapshot.Pods.Store(pod.UID, *pod) + entry := resolver.resolvePodDescriptor(pod) + for _, podIp := range pod.Status.PodIPs { + resolver.storeWorkloadsIP(podIp.IP, &entry) + podWorkload := Workload{ + Name: pod.Name, + Namespace: pod.Namespace, + Kind: "pod", + } + resolver.storePodsIP(podIp.IP, &podWorkload) + } + case watch.Modified: + pod, ok := podEvent.Object.(*v1.Pod) + if !ok { + return + } + resolver.snapshot.Pods.Store(pod.UID, *pod) + entry := resolver.resolvePodDescriptor(pod) + for _, podIp := range pod.Status.PodIPs { + resolver.storeWorkloadsIP(podIp.IP, &entry) + podWorkload := Workload{ + Name: pod.Name, + Namespace: pod.Namespace, + Kind: "pod", + } + resolver.storePodsIP(podIp.IP, &podWorkload) + } + 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) 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) + for _, nodeAddress := range node.Status.Addresses { + resolver.storeWorkloadsIP(nodeAddress.Address, &Workload{ + Name: node.Name, + Namespace: "node", + Kind: "node", + }) + } + case watch.Deleted: + if val, ok := nodeEvent.Object.(*v1.Node); ok { + resolver.snapshot.Nodes.Delete(val.UID) + } + } +} + +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) 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) + } + } +} + +func (resolver *K8sIPResolver) handleServicesWatchEvent(servicesEvent *watch.Event) { + switch servicesEvent.Type { + case watch.Added, watch.Modified: + service, ok := servicesEvent.Object.(*v1.Service) + 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", + } + + // 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) + } + case watch.Deleted: + if val, ok := deploymentsEvent.Object.(*appsv1.Deployment); ok { + resolver.snapshot.Deployments.Delete(val.UID) + } + } +} + +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) getResolvedClusterSnapshot() error { + log.Printf("Generating full cluster snapshot") + err := resolver.getFullClusterSnapshot() + if err != nil { + return err + } + resolver.updateIpMapping() + log.Printf("Generated full cluster snapshot") + return nil +} + +// iterate the API for initial coverage of the cluster's state +func (resolver *K8sIPResolver) getFullClusterSnapshot() error { + pods, err := resolver.clientset.CoreV1().Pods("").List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting pods, aborting snapshot update") + } + log.Printf("loaded pods data %d", pods.Size()) + + for _, pod := range pods.Items { + resolver.snapshot.Pods.Store(pod.UID, pod) + } + + nodes, err := resolver.clientset.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{}) + if err != nil { + return errors.New("error getting nodes, aborting snapshot update") + } + for _, node := range nodes.Items { + 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", + } + + // 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.Pods.Range(func(key, value any) bool { + pod, ok := value.(v1.Pod) + if !ok { + log.Printf("Type confusion in pods map") + return true // continue + } + entry := resolver.resolvePodDescriptor(&pod) + for _, podIp := range pod.Status.PodIPs { + // if ip is already in the map, override only if current pod is running + resolver.storeWorkloadsIP(podIp.IP, &entry) + podWorkload := Workload{ + Name: pod.Name, + Namespace: pod.Namespace, + Kind: "pod", + } + resolver.storePodsIP(podIp.IP, &podWorkload) + } + return true + }) + + resolver.snapshot.Nodes.Range(func(key any, value any) bool { + node, ok := value.(v1.Node) + if !ok { + log.Printf("Type confusion in nodes map") + return true // continue + } + for _, nodeAddress := range node.Status.Addresses { + workload := Workload{ + Name: node.Name, + Namespace: "node", + Kind: "node", + } + resolver.storeWorkloadsIP(nodeAddress.Address, &workload) + } + return true + }) +} + +func (resolver *K8sIPResolver) storeWorkloadsIP(ip string, newWorkload *Workload) { + // we want to override existing workload, unless the existing workload is a node and the new one isn't + val, ok := resolver.ipsMap.Load(ip) + if ok { + existingWorkload, ok := val.(Workload) + if ok { + if existingWorkload.Kind == "node" && newWorkload.Kind != "node" { + return + } + } + } + resolver.ipsMap.Store(ip, *newWorkload) +} + +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) +} + +func (resolver *K8sIPResolver) resolvePodDescriptor(pod *v1.Pod) Workload { + existing, ok := resolver.snapshot.PodDescriptors.Load(pod.UID) + if ok { + result, ok := existing.(Workload) + if ok { + return result + } + } + var err error + 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) + } + } + result := Workload{ + Name: name, + Namespace: namespace, + Kind: kind, + } + if err == nil { + resolver.snapshot.PodDescriptors.Store(pod.UID, result) + } + return result +} diff --git a/common/ip_resolver_test.go b/common/ip_resolver_test.go new file mode 100644 index 00000000..b938a269 --- /dev/null +++ b/common/ip_resolver_test.go @@ -0,0 +1,895 @@ +package common + +import ( + "log" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/watch" + testclient "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +type podDescriptor struct { + Name string + Namespace string + IP string + Phase v1.PodPhase + UID types.UID + Controller *workloadResourceDescriptor +} + +type nodeDescriptor struct { + Name string + IP string + UID types.UID +} + +type workloadResourceDescriptor struct { + Name string + Namespace string + UID types.UID + Kind string +} + +func (desc *workloadResourceDescriptor) CreateObject() runtime.Object { + switch desc.Kind { + case "Deployment": + { + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + case "ReplicaSet": + { + return &appsv1.ReplicaSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + case "DaemonSet": + { + return &appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + case "StatefulSet": + { + return &appsv1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + case "Job": + { + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + case "Service": + { + return &v1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + case "CronJob": + { + return &batchv1.CronJob{ + ObjectMeta: metav1.ObjectMeta{ + Name: desc.Name, + Namespace: desc.Namespace, + UID: desc.UID, + }, + } + } + } + return nil +} + +func generatePod(pod podDescriptor) runtime.Object { + newPod := v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: pod.Name, + Namespace: pod.Namespace, + UID: pod.UID, + }, + Status: v1.PodStatus{ + PodIP: pod.IP, + PodIPs: []v1.PodIP{ + {IP: pod.IP}, + }, + }, + } + if pod.Controller != nil { + newTrue := new(bool) + *newTrue = true + ref := metav1.OwnerReference{ + Kind: pod.Controller.Kind, + Name: pod.Controller.Name, + UID: pod.Controller.UID, + Controller: newTrue, + } + newPod.OwnerReferences = append(newPod.OwnerReferences, ref) + } + return &newPod + +} + +func generateWorkloadResource(desc workloadResourceDescriptor) runtime.Object { + return desc.CreateObject() +} + +func generateNode(node nodeDescriptor) runtime.Object { + return &v1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: node.Name, + UID: node.UID, + }, + Status: v1.NodeStatus{ + Addresses: []v1.NodeAddress{ + { + Type: "InternalIP", + Address: node.IP, + }, + }, + }, + } +} + +func generateClusterObjects(pods []podDescriptor, workloadsResources []workloadResourceDescriptor, nodes []nodeDescriptor) []runtime.Object { + result := make([]runtime.Object, 0, len(pods)+len(workloadsResources)+len(nodes)) + for _, pod := range pods { + newPod := generatePod(pod) + result = append(result, newPod) + } + for _, desc := range workloadsResources { + result = append(result, generateWorkloadResource(desc)) + } + for _, node := range nodes { + result = append(result, generateNode(node)) + } + return result +} + +type testStep struct { + shouldWait bool + newPods []podDescriptor + newNodes []nodeDescriptor + newWorkloadResource []workloadResourceDescriptor + modifiedPods []podDescriptor + modifiedNodes []nodeDescriptor + modifiedWorkloadResources []workloadResourceDescriptor + expectedResolves map[string]Workload +} + +type testScenario struct { + description string + initialState testStep + updateSteps []testStep +} + +type fakeWatchers struct { + nodesWatcher *watch.FakeWatcher + podsWatcher *watch.FakeWatcher + deploymentsWatcher *watch.FakeWatcher + replicasetsWatcher *watch.FakeWatcher + daemonsetsWatcher *watch.FakeWatcher + statefulsetsWatcher *watch.FakeWatcher + jobsWatcher *watch.FakeWatcher + servicesWatcher *watch.FakeWatcher + cronjobsWatcher *watch.FakeWatcher +} + +func createPrependWatchers(clientset *testclient.Clientset) fakeWatchers { + watchers := fakeWatchers{ + nodesWatcher: watch.NewFake(), + podsWatcher: watch.NewFake(), + deploymentsWatcher: watch.NewFake(), + replicasetsWatcher: watch.NewFake(), + daemonsetsWatcher: watch.NewFake(), + statefulsetsWatcher: watch.NewFake(), + jobsWatcher: watch.NewFake(), + servicesWatcher: watch.NewFake(), + cronjobsWatcher: watch.NewFake(), + } + clientset.PrependWatchReactor("nodes", k8stesting.DefaultWatchReactor(watchers.nodesWatcher, nil)) + clientset.PrependWatchReactor("pods", k8stesting.DefaultWatchReactor(watchers.podsWatcher, nil)) + clientset.PrependWatchReactor("deployments", k8stesting.DefaultWatchReactor(watchers.deploymentsWatcher, nil)) + clientset.PrependWatchReactor("replicasets", k8stesting.DefaultWatchReactor(watchers.replicasetsWatcher, nil)) + clientset.PrependWatchReactor("daemonsets", k8stesting.DefaultWatchReactor(watchers.daemonsetsWatcher, nil)) + clientset.PrependWatchReactor("statefulsets", k8stesting.DefaultWatchReactor(watchers.statefulsetsWatcher, nil)) + clientset.PrependWatchReactor("jobs", k8stesting.DefaultWatchReactor(watchers.jobsWatcher, nil)) + clientset.PrependWatchReactor("services", k8stesting.DefaultWatchReactor(watchers.servicesWatcher, nil)) + clientset.PrependWatchReactor("cronjobs", k8stesting.DefaultWatchReactor(watchers.cronjobsWatcher, nil)) + return watchers +} + +func addObject(watchers fakeWatchers, obj runtime.Object, kind string) { + switch kind { + case "Pod": + { + watchers.podsWatcher.Add(obj) + } + case "node": + { + watchers.nodesWatcher.Add(obj) + } + case "Deployment": + { + watchers.deploymentsWatcher.Add(obj) + } + case "ReplicaSet": + { + watchers.replicasetsWatcher.Add(obj) + } + case "DaemonSet": + { + watchers.daemonsetsWatcher.Add(obj) + } + case "StatefulSet": + { + watchers.statefulsetsWatcher.Add(obj) + } + case "Job": + { + watchers.jobsWatcher.Add(obj) + } + case "Service": + { + watchers.servicesWatcher.Add(obj) + } + case "CronJob": + { + watchers.cronjobsWatcher.Add(obj) + } + } +} + +func modifyObject(watchers fakeWatchers, obj runtime.Object, kind string) { + switch kind { + case "Pod": + { + watchers.podsWatcher.Modify(obj) + } + case "node": + { + watchers.nodesWatcher.Modify(obj) + } + case "Deployment": + { + watchers.deploymentsWatcher.Modify(obj) + } + case "ReplicaSet": + { + watchers.replicasetsWatcher.Modify(obj) + } + case "DaemonSet": + { + watchers.daemonsetsWatcher.Modify(obj) + } + case "StatefulSet": + { + watchers.statefulsetsWatcher.Modify(obj) + } + case "Job": + { + watchers.jobsWatcher.Modify(obj) + } + case "Service": + { + watchers.servicesWatcher.Modify(obj) + } + case "CronJob": + { + watchers.cronjobsWatcher.Modify(obj) + } + default: + { + log.Printf("unhandled kind %v", kind) + } + } +} + +func runTest(t *testing.T, test testScenario) { + assert := assert.New(t) + // Arrange 1: mocks and initial state + originalObjs := generateClusterObjects(test.initialState.newPods, test.initialState.newWorkloadResource, test.initialState.newNodes) + fakeClient := testclient.NewSimpleClientset(originalObjs...) + fakeWatchers := createPrependWatchers(fakeClient) + + resolver, err := NewK8sIPResolver(fakeClient, false) + assert.NoError(err) + + // Act 1: process initial state + err = resolver.StartWatching() + assert.NoError(err) + + // Assert 1: resolve and compare to expected, original state + for ipToCheck, expectedWorkload := range test.initialState.expectedResolves { + resultWorkload := resolver.ResolveIP(ipToCheck) + assert.Equal(expectedWorkload, resultWorkload) + } + + for _, step := range test.updateSteps { + // Arrange 2+n: update the state via watchers + for _, newPod := range step.newPods { + podObj := generatePod(newPod) + addObject(fakeWatchers, podObj, "Pod") + } + for _, newWorkloadResource := range step.newWorkloadResource { + obj := generateWorkloadResource(newWorkloadResource) + addObject(fakeWatchers, obj, newWorkloadResource.Kind) + } + for _, newNode := range step.newNodes { + obj := generateNode(newNode) + addObject(fakeWatchers, obj, "node") + } + for _, modifiedPod := range step.modifiedPods { + podObj := generatePod(modifiedPod) + modifyObject(fakeWatchers, podObj, "Pod") + } + for _, modifiedWorkloadResource := range step.newWorkloadResource { + obj := generateWorkloadResource(modifiedWorkloadResource) + modifyObject(fakeWatchers, obj, modifiedWorkloadResource.Kind) + } + for _, modifiedNode := range step.modifiedNodes { + obj := generateNode(modifiedNode) + modifyObject(fakeWatchers, obj, "node") + } + + if step.shouldWait { + time.Sleep(1 * time.Second) + } + + // Act+Assert 2+n + for ipToResolve, expectedWorkload := range step.expectedResolves { + assert.Equal(expectedWorkload, resolver.ResolveIP(ipToResolve)) + } + + } +} + +var testDeployment = workloadResourceDescriptor{"deployment1", "namespaceA", types.UID(uuid.NewString()), "Deployment"} +var testReplicaSet = workloadResourceDescriptor{"replicaset1", "namespaceA", types.UID(uuid.NewString()), "ReplicaSet"} +var testDaemonSet = workloadResourceDescriptor{"daemonset1", "namespaceA", types.UID(uuid.NewString()), "DaemonSet"} +var testStatefulSet = workloadResourceDescriptor{"statefulset1", "namespaceA", types.UID(uuid.NewString()), "StatefulSet"} +var testJob = workloadResourceDescriptor{"job1", "namespaceA", types.UID(uuid.NewString()), "Job"} +var testCronjob = workloadResourceDescriptor{"cronjob1", "namespaceA", types.UID(uuid.NewString()), "CronJob"} + +func TestResolving(t *testing.T) { + var tests = []testScenario{ + { + description: "unsuccessful resolving result should be external", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.2": { + Name: "1.1.1.2", + Namespace: "external", + Kind: "external", + }, + }, + }, + }, + { + description: "initial snapshot 1 pod resolve to pod1", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + { + description: "initial snapshot 3 pods resolve to each pod", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + {"pod2", "namespaceA", "1.1.1.2", v1.PodRunning, types.UID(uuid.New().String()), nil}, + {"pod3", "namespaceA", "1.1.1.3", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.2": { + Name: "pod2", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.3": { + Name: "pod3", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + { + description: "empty initial 1 pod added resolve to pod", + initialState: testStep{ + shouldWait: false, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "1.1.1.1", + Namespace: "external", + Kind: "external", + }, + }, + }, + updateSteps: []testStep{ + { + shouldWait: true, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + }, + { + description: "empty initial 1 node added resolve to node", + initialState: testStep{ + shouldWait: false, + expectedResolves: map[string]Workload{ + "1.1.1.0": { + Name: "1.1.1.0", + Namespace: "external", + Kind: "external", + }, + }, + }, + updateSteps: []testStep{ + { + shouldWait: true, + newNodes: []nodeDescriptor{ + {"Node1", "1.1.1.0", types.UID(uuid.NewString())}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.0": { + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + }, + }, + }, + }, + { + description: "empty initial 1 node, 1 pod added resolve to each", + initialState: testStep{ + shouldWait: false, + expectedResolves: map[string]Workload{ + "1.1.1.0": { + Name: "1.1.1.0", + Namespace: "external", + Kind: "external", + }, + }, + }, + updateSteps: []testStep{ + { + shouldWait: true, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + newNodes: []nodeDescriptor{ + {"Node1", "1.1.1.0", types.UID(uuid.NewString())}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.0": { + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + }, + { + description: "1 pod changing ip resolve both ips to pod", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.2": { + Name: "1.1.1.2", + Namespace: "external", + Kind: "external", + }, + }, + }, + updateSteps: []testStep{ + { + shouldWait: true, + modifiedPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.2", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { // the resolver shouldn't delete old not-reused entries + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.2": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + }, + { + description: "1 pod changing ip old ip is reused resolve reused ip to new pod", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID("1"), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.2": { + Name: "1.1.1.2", + Namespace: "external", + Kind: "external", + }, + }, + }, + updateSteps: []testStep{ + { + shouldWait: false, + modifiedPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.2", v1.PodRunning, types.UID("1"), nil}, + }, + expectedResolves: map[string]Workload{}, + }, + { + shouldWait: true, + newPods: []podDescriptor{ + {"pod2", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod2", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.2": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + }, + { + description: "1 pod changing ip old ip is reused by node resolve ip to new node", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID("1"), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + "1.1.1.2": { + Name: "1.1.1.2", + Namespace: "external", + Kind: "external", + }, + }, + }, + updateSteps: []testStep{ + { + shouldWait: false, + modifiedPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.2", v1.PodRunning, types.UID("1"), nil}, + }, + expectedResolves: map[string]Workload{}, + }, + { + shouldWait: true, + newNodes: []nodeDescriptor{ + {"Node1", "1.1.1.1", types.UID(uuid.NewString())}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + "1.1.1.2": { + Name: "pod1", + Namespace: "namespaceA", + Kind: "pod", + }, + }, + }, + }, + }, + { + description: "1 node changing ip resolve both ips to node", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{}, + newNodes: []nodeDescriptor{ + {"Node1", "1.1.1.0", types.UID("1")}, + }, + expectedResolves: map[string]Workload{}, + }, + updateSteps: []testStep{ + { + shouldWait: true, + modifiedNodes: []nodeDescriptor{ + {"Node1", "1.1.2.0", types.UID("1")}, + }, + modifiedWorkloadResources: []workloadResourceDescriptor{}, + expectedResolves: map[string]Workload{ + "1.1.1.0": { // resolver isn't expected to delete old not-reused entries + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + "1.1.2.0": { + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + }, + }, + }, + }, + { + description: "1 node changing ip, reused by another node resolve reused ip to new node", + initialState: testStep{ + shouldWait: true, + newNodes: []nodeDescriptor{ + {"Node1", "1.1.1.0", types.UID("1")}, + }, + expectedResolves: map[string]Workload{}, + }, + updateSteps: []testStep{ + { + shouldWait: true, + modifiedNodes: []nodeDescriptor{ + {"Node1", "1.1.2.0", types.UID("1")}, + }, + expectedResolves: map[string]Workload{}, + }, + { + shouldWait: true, + newNodes: []nodeDescriptor{ + {"Node2", "1.1.1.0", types.UID("2")}, + }, + modifiedNodes: []nodeDescriptor{ + {"Node1", "1.1.2.0", types.UID("1")}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.0": { + Name: "Node2", + Namespace: "node", + Kind: "node", + }, + "1.1.2.0": { + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + }, + }, + }, + }, + { + description: "pod with hostip wont override node", + initialState: testStep{ + shouldWait: false, + newNodes: []nodeDescriptor{ + {"Node1", "1.1.1.0", types.UID(uuid.NewString())}, + }, + expectedResolves: map[string]Workload{}, + }, + updateSteps: []testStep{ + { + shouldWait: true, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.0", v1.PodRunning, types.UID(uuid.New().String()), nil}, + }, + expectedResolves: map[string]Workload{ + "1.1.1.0": { + Name: "Node1", + Namespace: "node", + Kind: "node", + }, + }, + }, + }, + }, + } + for _, test := range tests { + t.Run(test.description, func(t *testing.T) { + runTest(t, test) + }) + } +} + +func TestControllersResolving(t *testing.T) { + var controllersTests = []testScenario{ + { + description: "initial snapshot 1 pod controlled by deployment resolve to deployment", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.NewString()), &testDeployment}, + }, + newWorkloadResource: []workloadResourceDescriptor{testDeployment}, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: testDeployment.Name, + Namespace: testDeployment.Namespace, + Kind: testDeployment.Kind, + }, + }, + }, + }, + { + description: "initial snapshot 1 pod controlled by replicaset resolve to replicaset", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.NewString()), &testReplicaSet}, + }, + newWorkloadResource: []workloadResourceDescriptor{testReplicaSet}, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: testReplicaSet.Name, + Namespace: testReplicaSet.Namespace, + Kind: testReplicaSet.Kind, + }, + }, + }, + }, + { + description: "initial snapshot 1 pod controlled by daemonset resolve to daemonset", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.NewString()), &testDaemonSet}, + }, + newWorkloadResource: []workloadResourceDescriptor{testDaemonSet}, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: testDaemonSet.Name, + Namespace: testDaemonSet.Namespace, + Kind: testDaemonSet.Kind, + }, + }, + }, + }, + { + description: "initial snapshot 1 pod controlled by statefulset resolve to statefulset", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.NewString()), &testStatefulSet}, + }, + newWorkloadResource: []workloadResourceDescriptor{testStatefulSet}, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: testStatefulSet.Name, + Namespace: testStatefulSet.Namespace, + Kind: testStatefulSet.Kind, + }, + }, + }, + }, + { + description: "initial snapshot 1 pod controlled by job resolve to job", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.NewString()), &testJob}, + }, + newWorkloadResource: []workloadResourceDescriptor{testJob}, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: testJob.Name, + Namespace: testJob.Namespace, + Kind: testJob.Kind, + }, + }, + }, + }, + { + description: "initial snapshot 1 pod controlled by cronjob resolve to cronjob", + initialState: testStep{ + shouldWait: false, + newPods: []podDescriptor{ + {"pod1", "namespaceA", "1.1.1.1", v1.PodRunning, types.UID(uuid.NewString()), &testCronjob}, + }, + newWorkloadResource: []workloadResourceDescriptor{testCronjob}, + expectedResolves: map[string]Workload{ + "1.1.1.1": { + Name: testCronjob.Name, + Namespace: testCronjob.Namespace, + Kind: testCronjob.Kind, + }, + }, + }, + }, + } + for _, test := range controllersTests { + t.Run(test.description, func(t *testing.T) { + runTest(t, test) + }) + } +} diff --git a/containers/container.go b/containers/container.go index a6ed463a..d753d1ca 100644 --- a/containers/container.go +++ b/containers/container.go @@ -63,17 +63,23 @@ func (p *LogParser) Stop() { } type AddrPair struct { - src netaddr.IPPort - dst netaddr.IPPort + src netaddr.IPPort + dst netaddr.IPPort + srcWorkload common.Workload + dstWorkload common.Workload + actualDestWorkload common.Workload } type ActiveConnection struct { - Dest netaddr.IPPort - ActualDest netaddr.IPPort - Pid uint32 - Fd uint64 - Timestamp uint64 - Closed time.Time + Dest netaddr.IPPort + ActualDest netaddr.IPPort + srcWorkload common.Workload + dstWorkload common.Workload + actialDestWorkload common.Workload + Pid uint32 + Fd uint64 + Timestamp uint64 + Closed time.Time http2Parser *l7.Http2Parser postgresParser *l7.PostgresParser @@ -129,10 +135,11 @@ type Container struct { lock sync.RWMutex - done chan struct{} + done chan struct{} + ip_resolver IPResolver } -func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, hostConntrack *Conntrack, pid uint32) (*Container, error) { +func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, hostConntrack *Conntrack, pid uint32, ip_resolver IPResolver) (*Container, error) { netNs, err := proc.GetNetNs(pid) if err != nil { return nil, err @@ -164,7 +171,8 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, host hostConntrack: hostConntrack, - done: make(chan struct{}), + done: make(chan struct{}), + ip_resolver: ip_resolver, } for _, n := range md.networks { @@ -286,13 +294,28 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } for d, count := range c.connectsSuccessful { - ch <- counter(metrics.NetConnectsSuccessful, float64(count), d.src.String(), d.dst.String()) + workload_src := d.srcWorkload + workload_dest := d.dstWorkload + actualDestWorkload := d.actualDestWorkload + if d.srcWorkload.Name == "" || len(d.srcWorkload.Name) == 0 { + workload_src = c.ip_resolver.ResolveIP(d.src.IP().String()) + workload_dest = c.ip_resolver.ResolveIP(d.dst.IP().String()) + } + ch <- counter(metrics.NetConnectsSuccessful, float64(count), d.src.String(), d.dst.String(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) } for dst, count := range c.connectsFailed { - ch <- counter(metrics.NetConnectsFailed, float64(count), dst.String()) + workload := c.ip_resolver.ResolveIP(dst.IP().String()) + ch <- counter(metrics.NetConnectsFailed, float64(count), dst.String(), workload.Name, workload.Namespace, workload.Kind, workload.Name, workload.Namespace, workload.Kind) } for d, count := range c.retransmits { - ch <- counter(metrics.NetRetransmits, float64(count), d.src.String(), d.dst.String()) + workload_src := d.srcWorkload + workload_dest := d.dstWorkload + + if d.srcWorkload.Name == "" || len(d.srcWorkload.Name) == 0 { + workload_src = c.ip_resolver.ResolveIP(d.src.IP().String()) + workload_dest = c.ip_resolver.ResolveIP(d.dst.IP().String()) + } + ch <- counter(metrics.NetRetransmits, float64(count), d.src.String(), d.dst.String(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind) } connections := map[AddrPair]int{} @@ -300,15 +323,20 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { if !conn.Closed.IsZero() { continue } - connections[AddrPair{src: addrPair.dst, dst: conn.ActualDest}]++ + workload_src := c.ip_resolver.ResolveIP(addrPair.dst.IP().String()) + workload_dest := c.ip_resolver.ResolveIP(conn.ActualDest.IP().String()) + actualDestWorkload := c.ip_resolver.ResolveActualIP(conn.ActualDest.IP().String()) + connections[AddrPair{src: addrPair.dst, dst: conn.ActualDest, srcWorkload: workload_src, dstWorkload: workload_dest, actualDestWorkload: actualDestWorkload}]++ } for d, count := range connections { - ch <- gauge(metrics.NetConnectionsActive, float64(count), d.src.String(), d.dst.String()) + ch <- gauge(metrics.NetConnectionsActive, float64(count), d.src.String(), d.dst.String(), d.srcWorkload.Name, d.srcWorkload.Namespace, d.srcWorkload.Kind, d.dstWorkload.Name, d.dstWorkload.Namespace, d.dstWorkload.Kind, d.actualDestWorkload.Name, d.actualDestWorkload.Namespace, d.actualDestWorkload.Kind) } for source, p := range c.logParsers { for _, c := range p.parser.GetCounters() { - ch <- counter(metrics.LogMessages, float64(c.Messages), source, c.Level.String(), c.Hash, c.Sample) + if c.Level == logparser.LevelCritical || c.Level == logparser.LevelError { + ch <- counter(metrics.LogMessages, float64(c.Messages), source, c.Level.String(), c.Hash, c.Sample) + } } } @@ -506,15 +534,22 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP if failed { c.connectsFailed[dst]++ } else { - c.connectsSuccessful[AddrPair{src: dst, dst: *actualDst}]++ + srcWorkload := c.ip_resolver.ResolveIP(src.IP().String()) + dstWorkload := c.ip_resolver.ResolveIP(dst.IP().String()) + actialDestWorkload := c.ip_resolver.ResolveActualIP(actualDst.IP().String()) + c.connectsSuccessful[AddrPair{src: dst, dst: *actualDst, srcWorkload: srcWorkload, + dstWorkload: dstWorkload, actualDestWorkload: actialDestWorkload}]++ connection := &ActiveConnection{ - Dest: dst, - ActualDest: *actualDst, - Pid: pid, - Fd: fd, - Timestamp: timestamp, - } - c.connectionsActive[AddrPair{src: src, dst: dst}] = connection + Dest: dst, + ActualDest: *actualDst, + Pid: pid, + Fd: fd, + Timestamp: timestamp, + srcWorkload: srcWorkload, + dstWorkload: dstWorkload, + actialDestWorkload: actialDestWorkload, + } + c.connectionsActive[AddrPair{src: src, dst: dst, srcWorkload: srcWorkload, dstWorkload: dstWorkload, actualDestWorkload: actialDestWorkload}] = connection c.connectionsByPidFd[PidFd{Pid: pid, Fd: fd}] = connection } c.connectLastAttempt[dst] = time.Now() @@ -572,7 +607,8 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R if timestamp != 0 && conn.Timestamp != timestamp { return } - stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest, r) + + stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest, r, conn.srcWorkload, conn.dstWorkload, conn.actialDestWorkload) trace := tracing.NewTrace(string(c.id), conn.ActualDest) switch r.Protocol { case l7.ProtocolHTTP: @@ -634,7 +670,9 @@ func (c *Container) onRetransmit(srcDst AddrPair) bool { if !ok { return false } - c.retransmits[AddrPair{src: srcDst.dst, dst: conn.ActualDest}]++ + src_workload := c.ip_resolver.ResolveIP(srcDst.dst.IP().String()) + dst_workload := c.ip_resolver.ResolveIP(conn.ActualDest.IP().String()) + c.retransmits[AddrPair{src: srcDst.dst, dst: conn.ActualDest, srcWorkload: src_workload, dstWorkload: dst_workload}]++ return true } diff --git a/containers/l7.go b/containers/l7.go index eb50db2b..f4782eea 100644 --- a/containers/l7.go +++ b/containers/l7.go @@ -3,6 +3,7 @@ package containers import ( "time" + "github.com/coroot/coroot-node-agent/common" "github.com/coroot/coroot-node-agent/ebpftracer/l7" "github.com/prometheus/client_golang/prometheus" "inet.af/netaddr" @@ -36,7 +37,7 @@ func (m *L7Metrics) observe(status, method string, duration time.Duration) { type L7Stats map[l7.Protocol]map[AddrPair]*L7Metrics // protocol -> dst:actual_dst -> metrics -func (s L7Stats) get(protocol l7.Protocol, destination, actualDestination netaddr.IPPort, r *l7.RequestData) *L7Metrics { +func (s L7Stats) get(protocol l7.Protocol, destination, actualDestination netaddr.IPPort, r *l7.RequestData, srcWorkload common.Workload, dstWorkload common.Workload, actualDstWorkload common.Workload) *L7Metrics { if protocol == l7.ProtocolHTTP2 { protocol = l7.ProtocolHTTP } @@ -45,12 +46,23 @@ func (s L7Stats) get(protocol l7.Protocol, destination, actualDestination netadd protoStats = map[AddrPair]*L7Metrics{} s[protocol] = protoStats } - dest := AddrPair{src: destination, dst: actualDestination} + dest := AddrPair{src: destination, dst: actualDestination, srcWorkload: srcWorkload, dstWorkload: dstWorkload, actualDestWorkload: actualDstWorkload} m := protoStats[dest] if m == nil { m = &L7Metrics{} protoStats[dest] = m - constLabels := map[string]string{"destination": destination.String(), "actual_destination": actualDestination.String()} + constLabels := map[string]string{"destination": destination.String(), + "actual_destination": actualDestination.String(), + "destination_workload_kind": dstWorkload.Kind, + "destination_workload_name": dstWorkload.Name, + "destination_workload_namespace": dstWorkload.Namespace, + "src_kind": srcWorkload.Kind, + "src_workload_name": srcWorkload.Name, + "src_workload_namespace": srcWorkload.Namespace, + "actual_destination_workload_kind": actualDstWorkload.Kind, + "actual_destination_workload_name": actualDstWorkload.Name, + "actual_destination_workload_namespace": actualDstWorkload.Namespace, + } labels := []string{"status"} switch protocol { case l7.ProtocolRabbitmq, l7.ProtocolNats: @@ -59,6 +71,10 @@ func (s L7Stats) get(protocol l7.Protocol, destination, actualDestination netadd method, path := l7.ParseHttp(r.Payload) constLabels["path"] = path constLabels["method"] = method + hOpts := L7Latency[protocol] + m.Latency = prometheus.NewHistogram( + prometheus.HistogramOpts{Name: hOpts.Name, Help: hOpts.Help, ConstLabels: constLabels}, + ) default: hOpts := L7Latency[protocol] m.Latency = prometheus.NewHistogram( diff --git a/containers/metrics.go b/containers/metrics.go index e3f039d5..78111dee 100644 --- a/containers/metrics.go +++ b/containers/metrics.go @@ -70,10 +70,10 @@ var metrics = struct { DiskWriteBytes: metric("container_resources_disk_written_bytes_total", "Total number of bytes written to the disk by the container", "mount_point", "device", "volume"), NetListenInfo: metric("container_net_tcp_listen_info", "Listen address of the container", "listen_addr", "proxy"), - NetConnectsSuccessful: metric("container_net_tcp_successful_connects_total", "Total number of successful TCP connects", "destination", "actual_destination"), - NetConnectsFailed: metric("container_net_tcp_failed_connects_total", "Total number of failed TCP connects", "destination"), - NetConnectionsActive: metric("container_net_tcp_active_connections", "Number of active outbound connections used by the container", "destination", "actual_destination"), - NetRetransmits: metric("container_net_tcp_retransmits_total", "Total number of retransmitted TCP segments", "destination", "actual_destination"), + NetConnectsSuccessful: metric("container_net_tcp_successful_connects_total", "Total number of successful TCP connects", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), + NetConnectsFailed: metric("container_net_tcp_failed_connects_total", "Total number of failed TCP connects", "destination", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), + NetConnectionsActive: metric("container_net_tcp_active_connections", "Number of active outbound connections used by the container", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), + NetRetransmits: metric("container_net_tcp_retransmits_total", "Total number of retransmitted TCP segments", "destination", "actual_destination", "src_workload_name", "src_workload_namespace", "src_workload_kind", "destination_workload_name", "destination_workload_namespace", "destination_workload_kind", "actual_destination_workload_name", "actual_destination_workload_namespace", "actual_destination_workload_kind"), NetLatency: metric("container_net_latency_seconds", "Round-trip time between the container and a remote IP", "destination_ip"), LogMessages: metric("container_log_messages_total", "Number of messages grouped by the automatically extracted repeated pattern", "source", "level", "pattern_hash", "sample"), diff --git a/containers/registry.go b/containers/registry.go index 65afc4bf..fcef96d6 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -31,6 +31,13 @@ type ProcessInfo struct { StartedAt time.Time } +type IPResolver interface { + ResolveIP(string) common.Workload + ResolveActualIP(string) common.Workload + StartWatching() error + StopWatching() +} + type Registry struct { reg prometheus.Registerer @@ -44,9 +51,10 @@ type Registry struct { containersByPid map[uint32]*Container processInfoCh chan<- ProcessInfo + ip_resolver IPResolver } -func NewRegistry(reg prometheus.Registerer, kernelVersion string, processInfoCh chan<- ProcessInfo) (*Registry, error) { +func NewRegistry(reg prometheus.Registerer, kernelVersion string, processInfoCh chan<- ProcessInfo, ip_resolver *common.K8sIPResolver) (*Registry, error) { ns, err := proc.GetSelfNetNs() if err != nil { return nil, err @@ -100,7 +108,8 @@ func NewRegistry(reg prometheus.Registerer, kernelVersion string, processInfoCh processInfoCh: processInfoCh, - tracer: ebpftracer.NewTracer(kernelVersion, *flags.DisableL7Tracing), + tracer: ebpftracer.NewTracer(kernelVersion, *flags.DisableL7Tracing), + ip_resolver: ip_resolver, } go r.handleEvents(r.events) @@ -303,7 +312,7 @@ func (r *Registry) getOrCreateContainer(pid uint32) *Container { r.containersByCgroupId[cg.Id] = c return c } - c, err := NewContainer(id, cg, md, r.hostConntrack, pid) + c, err := NewContainer(id, cg, md, r.hostConntrack, pid, r.ip_resolver) if err != nil { klog.Warningf("failed to create container pid=%d cg=%s id=%s: %s", pid, cg.Id, id, err) return nil diff --git a/flags/flags.go b/flags/flags.go index a9217b35..2e0833c1 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -32,7 +32,8 @@ var ( ApiKey = kingpin.Flag("api-key", "Coroot API key").Envar("API_KEY").String() ScrapeInterval = kingpin.Flag("scrape-interval", "How often to gather metrics from the agent").Default("15s").Envar("SCRAPE_INTERVAL").Duration() - WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/coroot-node-agent").Envar("WAL_DIR").String() + WalDir = kingpin.Flag("wal-dir", "Path to where the agent stores data (e.g. the metrics Write-Ahead Log)").Default("/tmp/coroot-node-agent").Envar("WAL_DIR").String() + ResolveDns = kingpin.Flag("resolve-dns", "should resolve DNS").Default("true").Envar("RESOLVE_DNS").Bool() ) func GetString(fl *string) string { diff --git a/go.mod b/go.mod index e4d61461..a7a590d3 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,9 @@ require ( github.com/docker/docker v25.0.0+incompatible github.com/florianl/go-conntrack v0.3.0 github.com/go-kit/log v0.2.1 + github.com/google/uuid v1.5.0 github.com/grafana/pyroscope/ebpf v0.4.1 + github.com/hashicorp/golang-lru/v2 v2.0.5 github.com/jpillora/backoff v1.0.0 github.com/mdlayher/taskstats v0.0.0-20230712191918-387b3d561d14 github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417 @@ -40,6 +42,9 @@ require ( gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/yaml.v2 v2.4.0 inet.af/netaddr v0.0.0-20230525184311-b8eac61e914a + k8s.io/api v0.28.6 + k8s.io/apimachinery v0.28.6 + k8s.io/client-go v0.28.6 k8s.io/klog/v2 v2.120.1 ) @@ -73,6 +78,8 @@ require ( github.com/docker/go-connections v0.4.0 // indirect github.com/docker/go-events v0.0.0-20190806004212-e31b211e4f1c // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/emicklei/go-restful/v3 v3.11.0 // indirect + github.com/evanphx/json-patch v5.6.0+incompatible // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.7.0 // indirect github.com/go-logfmt/logfmt v0.6.0 // indirect @@ -96,13 +103,13 @@ require ( github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect github.com/golang/protobuf v1.5.3 // indirect github.com/golang/snappy v0.0.4 // indirect + github.com/google/gnostic-models v0.6.8 // indirect github.com/google/go-cmp v0.6.0 // indirect + github.com/google/gofuzz v1.2.0 // indirect github.com/google/pprof v0.0.0-20240117000934-35fc243c5815 // indirect - github.com/google/uuid v1.5.0 // indirect github.com/grafana/regexp v0.0.0-20221123153739-15dc172cd2db // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect github.com/hashicorp/go-version v1.6.0 // indirect - github.com/hashicorp/golang-lru/v2 v2.0.5 // indirect github.com/hashicorp/hcl v1.0.0 // indirect github.com/ianlancetaylor/demangle v0.0.0-20230524184225-eabc099b10ab // indirect github.com/inconshreveable/mousetrap v1.0.1 // indirect @@ -125,6 +132,7 @@ require ( github.com/moby/term v0.5.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f // indirect github.com/oklog/ulid v1.3.1 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect @@ -169,6 +177,7 @@ require ( golang.org/x/exp v0.0.0-20240119083558-1b970713d09a // indirect golang.org/x/oauth2 v0.16.0 // indirect golang.org/x/sync v0.6.0 // indirect + golang.org/x/term v0.16.0 // indirect golang.org/x/text v0.14.0 // indirect golang.org/x/tools v0.17.0 // indirect google.golang.org/appengine v1.6.8 // indirect @@ -176,11 +185,14 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20240116215550-a9fa1716bcac // indirect google.golang.org/grpc v1.61.0 // indirect google.golang.org/protobuf v1.33.0 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect - k8s.io/apimachinery v0.28.6 // indirect - k8s.io/client-go v0.28.6 // indirect + k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect + sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect + sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect + sigs.k8s.io/yaml v1.3.0 // indirect ) replace ( diff --git a/go.sum b/go.sum index f4ef141f..81bc6f1c 100644 --- a/go.sum +++ b/go.sum @@ -352,7 +352,6 @@ github.com/edsrzf/mmap-go v1.1.0/go.mod h1:19H/e8pUPLicwkyNgOykDXkJ9F0MHE+Z52B8E github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkgs4iZTTu3o/KG3Itv/qCCa8VVMlb3i9OVuzc= github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs= github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs= -github.com/emicklei/go-restful v2.16.0+incompatible h1:rgqiKNjTnFQA6kkhFe16D8epTksy9HQ1MyrbDXSdYhM= github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g= github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= @@ -367,6 +366,8 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7 github.com/envoyproxy/protoc-gen-validate v1.0.4 h1:gVPz/FMfvh57HdSJQyvBtF00j8JU4zdyUgIUNhlgg0A= github.com/envoyproxy/protoc-gen-validate v1.0.4/go.mod h1:qys6tmnRsYrQqIhm2bvKZH4Blx/1gTIZ2UKVY1M+Yew= github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= +github.com/evanphx/json-patch v5.6.0+incompatible h1:jBYDEEiFBPxA0v50tFdvOzQQTCvpL6mnFh5mB2/l16U= +github.com/evanphx/json-patch v5.6.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk= github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb h1:IT4JYU7k4ikYg1SCxNI1/Tieq/NFvh6dzLdgi7eu0tM= github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb/go.mod h1:bH6Xx7IW64qjjJq8M2u4dxNaBiDfKK+z/3eGDpXEQhc= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= @@ -462,6 +463,8 @@ github.com/go-openapi/validate v0.22.1/go.mod h1:rjnrwK57VJ7A8xqfpAOEKRH8yQSGUri github.com/go-resty/resty/v2 v2.11.0 h1:i7jMfNOJYMp69lq7qozJP+bjgzfAzeOhuGlyDrqxT/8= github.com/go-resty/resty/v2 v2.11.0/go.mod h1:iiP/OpA0CkcL3IGt1O0+/SIItFUbkkyw5BGXiVdTu+A= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= +github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI= +github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls= github.com/go-zookeeper/zk v1.0.3 h1:7M2kwOsc//9VeeFiPtf+uSJlVpU66x9Ba5+8XK7/TDg= github.com/go-zookeeper/zk v1.0.3/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw= github.com/gobuffalo/attrs v0.0.0-20190224210810-a9411de4debd/go.mod h1:4duuawTqi2wkkpB4ePgWMaai6/Kc6WEz83bhFwpHzj0= @@ -851,6 +854,9 @@ github.com/onsi/ginkgo v1.10.3/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+ github.com/onsi/ginkgo v1.11.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.12.0/go.mod h1:oUhWkIvk5aDxtKvDDuw8gItl8pKl42LzjC9KZE0HfGg= github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk= +github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= +github.com/onsi/ginkgo/v2 v2.9.4 h1:xR7vG4IXt5RWx6FfIjyAtsoMAtnc3C/rFXBBd2AjZwE= +github.com/onsi/ginkgo/v2 v2.9.4/go.mod h1:gCQYp2Q+kSoIj7ykSVb9nskRSsR6PUj4AiLywzIhbKM= github.com/onsi/gomega v0.0.0-20151007035656-2152b45fa28a/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA= github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA= github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= @@ -858,6 +864,8 @@ github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1Cpa github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY= github.com/onsi/gomega v1.9.0/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA= github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc= +github.com/onsi/gomega v1.27.6 h1:ENqfyGeS5AX/rlXDd/ETokDz93u0YufY1Pgxuy/PvWE= +github.com/onsi/gomega v1.27.6/go.mod h1:PIQNjfQwkP3aQAH7lf7j87O/5FiNr+ZR8+ipb+qQlhg= github.com/opencontainers/go-digest v0.0.0-20170106003457-a6d0ee40d420/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s= github.com/opencontainers/go-digest v0.0.0-20180430190053-c9281466c8b2/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s= github.com/opencontainers/go-digest v1.0.0-rc1/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s= diff --git a/main.go b/main.go index a6ca148b..4c2dd2f8 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "log" "net/http" _ "net/http/pprof" "os" @@ -22,6 +23,8 @@ import ( "golang.org/x/mod/semver" "golang.org/x/sys/unix" "golang.org/x/time/rate" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" "k8s.io/klog/v2" ) @@ -95,12 +98,37 @@ func whitelistNodeExternalNetworks() { } } +func getClientSet() (*kubernetes.Clientset, error) { + config, err := rest.InClusterConfig() + if err != nil { + return nil, err + } + + clientset, err := kubernetes.NewForConfig(config) + if err != nil { + return nil, err + } + return clientset, nil +} + func main() { klog.LogToStderr(false) klog.SetOutput(&RateLimitedLogOutput{limiter: rate.NewLimiter(rate.Limit(*flags.LogPerSecond), *flags.LogBurst)}) klog.Infoln("agent version:", version) + clientset, err := getClientSet() + if err != nil { + log.Fatalf("Error getting kubernetes clientset: %v", err) + } + resolver, err := common.NewK8sIPResolver(clientset, *flags.ResolveDns) + if err != nil { + log.Fatalf("Error creating resolver: %v", err) + } + err = resolver.StartWatching() + if err != nil { + log.Fatalf("Error watching cluster's state: %v", err) + } hostname, kv, err := uname() if err != nil { klog.Exitln("failed to get uname:", err) @@ -133,7 +161,7 @@ func main() { processInfoCh := profiling.Init(machineId, hostname) - cr, err := containers.NewRegistry(registerer, kv, processInfoCh) + cr, err := containers.NewRegistry(registerer, kv, processInfoCh, resolver) if err != nil { klog.Exitln(err) }