Merge pull request #18 from coroot/docker_swarm

Initial support for Docker Swarm

Author: def

containers/container.go

@@ -28,6 +28,10 @@ var (
 
 type ContainerID string
 
+type ContainerNetwork struct {
+	NetworkID string
+}
+
 type ContainerMetadata struct {
 	name        string
 	labels      map[string]string
@@ -36,6 +40,7 @@ type ContainerMetadata struct {
 	image       string
 	logDecoder  logparser.Decoder
 	hostListens map[string][]netaddr.IPPort
+	networks    map[string]ContainerNetwork
 }
 
 type Delays struct {
@@ -111,6 +116,7 @@ type Container struct {
 	isHostNs      bool
 	hostConntrack *Conntrack
 	nsConntrack   *Conntrack
+	lbConntracks  []*Conntrack
 
 	lock sync.RWMutex
 
@@ -151,6 +157,17 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, host
 		done: make(chan struct{}),
 	}
 
+	for _, n := range md.networks {
+		if nsHandle := FindNetworkLoadBalancerNs(n.NetworkID); nsHandle.IsOpen() {
+			if ct, err := NewConntrack(nsHandle); err != nil {
+				klog.Warningln(err)
+			} else {
+				c.lbConntracks = append(c.lbConntracks, ct)
+			}
+			_ = nsHandle.Close()
+		}
+	}
+
 	c.runLogParser("")
 
 	go func() {
@@ -173,6 +190,9 @@ func (c *Container) Close() {
 	for _, p := range c.logParsers {
 		p.Stop()
 	}
+	for _, ct := range c.lbConntracks {
+		_ = ct.Close()
+	}
 	if c.nsConntrack != nil {
 		_ = c.nsConntrack.Close()
 	}
@@ -477,6 +497,11 @@ func (c *Container) getActualDestination(pid uint32, src, dst netaddr.IPPort) (*
 	if actualDst := lookupCiliumConntrackTable(src, dst); actualDst != nil {
 		return actualDst, nil
 	}
+	for _, lb := range c.lbConntracks {
+		if actualDst := lb.GetActualDestination(src, dst); actualDst != nil {
+			return actualDst, nil
+		}
+	}
 	actualDst := c.hostConntrack.GetActualDestination(src, dst)
 	if actualDst != nil {
 		return actualDst, nil

containers/dockerd.go

@@ -7,7 +7,10 @@ import (
 	"github.com/coroot/coroot-node-agent/proc"
 	"github.com/coroot/logparser"
 	"github.com/docker/docker/client"
+	"github.com/vishvananda/netns"
 	"inet.af/netaddr"
+	"os"
+	"path"
 	"strings"
 	"time"
 )
@@ -51,6 +54,7 @@ func DockerdInspect(containerID string) (*ContainerMetadata, error) {
 		image:       c.Config.Image,
 		volumes:     map[string]string{},
 		hostListens: map[string][]netaddr.IPPort{},
+		networks:    map[string]ContainerNetwork{},
 	}
 	for _, m := range c.Mounts {
 		res.volumes[m.Destination] = common.ParseKubernetesVolumeSource(m.Source)
@@ -78,6 +82,30 @@ func DockerdInspect(containerID string) (*ContainerMetadata, error) {
 			}
 			res.hostListens["dockerd"] = s
 		}
+		for name, network := range c.NetworkSettings.Networks {
+			res.networks[name] = ContainerNetwork{
+				NetworkID: network.NetworkID,
+			}
+		}
 	}
 	return res, nil
 }
+
+func FindNetworkLoadBalancerNs(networkId string) netns.NsHandle {
+	basePath := "/run/docker/netns"
+	files, err := os.ReadDir(proc.HostPath(basePath))
+	if err != nil {
+		return -1
+	}
+	for _, f := range files {
+		if !f.Type().IsRegular() || !strings.HasPrefix(f.Name(), "lb_") {
+			continue
+		}
+		idPrefix := strings.Split(f.Name(), "_")[1]
+		if strings.HasPrefix(networkId, idPrefix) {
+			ns, _ := netns.GetFromPath(proc.HostPath(path.Join(basePath, f.Name())))
+			return ns
+		}
+	}
+	return -1
+}

containers/registry.go

@@ -13,6 +13,7 @@ import (
 	"k8s.io/klog/v2"
 	"os"
 	"regexp"
+	"strings"
 	"time"
 )
 
@@ -325,6 +326,17 @@ func calcId(cg *cgroup.Cgroup, md *ContainerMetadata) ContainerID {
 		}
 		return ContainerID(fmt.Sprintf("/k8s/%s/%s/%s", namespace, pod, name))
 	}
+	if taskNameParts := strings.SplitN(md.labels["com.docker.swarm.task.name"], ".", 3); len(taskNameParts) == 3 {
+		namespace := md.labels["com.docker.stack.namespace"]
+		service := md.labels["com.docker.swarm.service.name"]
+		if namespace != "" {
+			service = strings.TrimPrefix(service, namespace+"_")
+		}
+		if namespace == "" {
+			namespace = "_"
+		}
+		return ContainerID(fmt.Sprintf("/swarm/%s/%s/%s", namespace, service, taskNameParts[1]))
+	}
 	if md.name == "" { // should be "pure" dockerd container here
 		klog.Warningln("empty dockerd container name for:", cg.ContainerId)
 		return ""