Allow reachability across services on the same host

controller.go

@@ -660,6 +660,9 @@ func (c *controller) NewNetwork(networkType, name string, id string, options ...
 	}
 
 	joinCluster(network)
+	if !c.isDistributedControl() {
+		arrangeIngressFilterRule()
+	}
 
 	return network, nil
 }

service_linux.go

@@ -479,14 +479,19 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro
 	}
 
 	chainExists := iptables.ExistChain(ingressChain, iptables.Nat)
+	filterChainExists := iptables.ExistChain(ingressChain, iptables.Filter)
 
 	ingressOnce.Do(func() {
+		// Flush nat table and filter table ingress chain rules during init if it
+		// exists. It might contain stale rules from previous life.
 		if chainExists {
-			// Flush ingress chain rules during init if it
-			// exists. It might contain stale rules from
-			// previous life.
 			if err := iptables.RawCombinedOutput("-t", "nat", "-F", ingressChain); err != nil {
-				logrus.Errorf("Could not flush ingress chain rules during init: %v", err)
+				logrus.Errorf("Could not flush nat table ingress chain rules during init: %v", err)
+			}
+		}
+		if filterChainExists {
+			if err := iptables.RawCombinedOutput("-F", ingressChain); err != nil {
+				logrus.Errorf("Could not flush filter table ingress chain rules during init: %v", err)
 			}
 		}
 	})
@@ -497,10 +502,21 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro
 				return fmt.Errorf("failed to create ingress chain: %v", err)
 			}
 		}
+		if !filterChainExists {
+			if err := iptables.RawCombinedOutput("-N", ingressChain); err != nil {
+				return fmt.Errorf("failed to create filter table ingress chain: %v", err)
+			}
+		}
 
 		if !iptables.Exists(iptables.Nat, ingressChain, "-j", "RETURN") {
 			if err := iptables.RawCombinedOutput("-t", "nat", "-A", ingressChain, "-j", "RETURN"); err != nil {
-				return fmt.Errorf("failed to add return rule in ingress chain: %v", err)
+				return fmt.Errorf("failed to add return rule in nat table ingress chain: %v", err)
+			}
+		}
+
+		if !iptables.Exists(iptables.Filter, ingressChain, "-j", "RETURN") {
+			if err := iptables.RawCombinedOutput("-A", ingressChain, "-j", "RETURN"); err != nil {
+				return fmt.Errorf("failed to add return rule to filter table ingress chain: %v", err)
 			}
 		}
 
@@ -512,6 +528,12 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro
 			}
 		}
 
+		if !iptables.Exists(iptables.Filter, "FORWARD", "-j", ingressChain) {
+			if err := iptables.RawCombinedOutput("-I", "FORWARD", "-j", ingressChain); err != nil {
+				return fmt.Errorf("failed to add jump rule to %s in filter table forward chain: %v", ingressChain, err)
+			}
+		}
+
 		oifName, err := findOIFName(gwIP)
 		if err != nil {
 			return fmt.Errorf("failed to find gateway bridge interface name for %s: %v", gwIP, err)
@@ -544,6 +566,30 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro
 			}
 		}
 
+		// Filter table rules to allow a published service to be accessible in the local node from..
+		// 1) service tasks attached to other networks
+		// 2) unmanaged containers on bridge networks
+		rule := strings.Fields(fmt.Sprintf("%s %s -m state -p %s --sport %d --state ESTABLISHED,RELATED -j ACCEPT",
+			addDelOpt, ingressChain, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort))
+		if err := iptables.RawCombinedOutput(rule...); err != nil {
+			errStr := fmt.Sprintf("setting up rule failed, %v: %v", rule, err)
+			if !isDelete {
+				return fmt.Errorf("%s", errStr)
+			}
+			logrus.Warnf("%s", errStr)
+		}
+
+		rule = strings.Fields(fmt.Sprintf("%s %s -p %s --dport %d -j ACCEPT",
+			addDelOpt, ingressChain, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort))
+		if err := iptables.RawCombinedOutput(rule...); err != nil {
+			errStr := fmt.Sprintf("setting up rule failed, %v: %v", rule, err)
+			if !isDelete {
+				return fmt.Errorf("%s", errStr)
+			}
+
+			logrus.Warnf("%s", errStr)
+		}
+
 		if err := plumbProxy(iPort, isDelete); err != nil {
 			logrus.Warnf("failed to create proxy for port %d: %v", iPort.PublishedPort, err)
 		}
@@ -552,6 +598,22 @@ func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) erro
 	return nil
 }
 
+// In the filter table FORWARD chain first rule should be to jump to INGRESS-CHAIN
+// This chain has the rules to allow access to the published ports for swarm tasks
+// from local bridge networks and docker_gwbridge (ie:taks on other swarm netwroks)
+func arrangeIngressFilterRule() {
+	if iptables.ExistChain(ingressChain, iptables.Filter) {
+		if iptables.Exists(iptables.Filter, "FORWARD", "-j", ingressChain) {
+			if err := iptables.RawCombinedOutput("-D", "FORWARD", "-j", ingressChain); err != nil {
+				logrus.Warnf("failed to delete jump rule to ingressChain in filter table: %v", err)
+			}
+		}
+		if err := iptables.RawCombinedOutput("-I", "FORWARD", "-j", ingressChain); err != nil {
+			logrus.Warnf("failed to add jump rule to ingressChain in filter table: %v", err)
+		}
+	}
+}
+
 func findOIFName(ip net.IP) (string, error) {
 	nlh := ns.NlHandle()
 

service_unsupported.go

@@ -17,3 +17,6 @@ func (c *controller) rmServiceBinding(name, sid, nid, eid string, vip net.IP, in
 
 func (sb *sandbox) populateLoadbalancers(ep *endpoint) {
 }
+
+func arrangeIngressFilterRule() {
+}