diff --git a/common/ip_resolver.go b/common/ip_resolver.go index 99dcd0a6..c8fe4717 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -55,6 +55,7 @@ type K8sIPResolver struct { stopSignal chan bool shouldResolveDns bool dnsResolvedIps *lrucache.Cache[string, string] + podIpsMap sync.Map } type Workload struct { @@ -81,11 +82,40 @@ func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPRe 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) @@ -94,6 +124,13 @@ 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 + } + } host := ip if resolver.shouldResolveDns { @@ -334,6 +371,12 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { 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) @@ -344,6 +387,12 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { 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 { @@ -627,6 +676,12 @@ func (resolver *K8sIPResolver) updateIpMapping() { 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 }) @@ -663,6 +718,10 @@ func (resolver *K8sIPResolver) storeWorkloadsIP(ip string, newWorkload *Workload 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) { diff --git a/containers/container.go b/containers/container.go index 25f696f3..d753d1ca 100644 --- a/containers/container.go +++ b/containers/container.go @@ -63,21 +63,23 @@ func (p *LogParser) Stop() { } type AddrPair struct { - src netaddr.IPPort - dst netaddr.IPPort - srcWorkload common.Workload - dstWorkload common.Workload + src netaddr.IPPort + dst netaddr.IPPort + srcWorkload common.Workload + dstWorkload common.Workload + actualDestWorkload common.Workload } type ActiveConnection struct { - Dest netaddr.IPPort - ActualDest netaddr.IPPort - srcWorkload common.Workload - dstWorkload common.Workload - 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 @@ -292,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{} @@ -306,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) + } } } @@ -491,9 +513,6 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP if dst.IP().IsLoopback() && !p.isHostNs() { return } - srcWorkload := c.ip_resolver.ResolveIP(netaddr.IPPort(src).IP().String()) - dstWorkload := c.ip_resolver.ResolveIP(netaddr.IPPort(dst).IP().String()) - actualDst, err := c.getActualDestination(p, src, dst) if err != nil { if !common.IsNotExist(err) { @@ -515,17 +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, - srcWorkload: srcWorkload, - dstWorkload: dstWorkload, - } - c.connectionsActive[AddrPair{src: src, dst: dst, srcWorkload: srcWorkload, dstWorkload: dstWorkload}] = 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() @@ -584,7 +608,7 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R return } - stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest, r, conn.srcWorkload, conn.dstWorkload) + 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: @@ -646,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 716b044b..f4782eea 100644 --- a/containers/l7.go +++ b/containers/l7.go @@ -37,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, srcWorkload common.Workload, dstWorkload common.Workload) *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 } @@ -46,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(), "dest_kind": dstWorkload.Kind, "dest_workload_name": dstWorkload.Name, "dest_workload_namespace": dstWorkload.Namespace, "src_kind": srcWorkload.Kind, "src_workload_name": srcWorkload.Name, "src_workload_namespace": srcWorkload.Namespace} + 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: @@ -60,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 704d805e..fcef96d6 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -33,6 +33,7 @@ type ProcessInfo struct { type IPResolver interface { ResolveIP(string) common.Workload + ResolveActualIP(string) common.Workload StartWatching() error StopWatching() }