Move ToCIDR gen logic to k8s package

policy.Translator interface is created to allow policy repo to accept
custom policy rule manipulation logic

Signed-off-by: Maciej Kwiek <maciej@covalent.io>

daemon/k8s_watcher.go

@@ -521,7 +521,8 @@ func (d *Daemon) endpointAddFn(obj interface{}) {
 
 	svc, ok := d.loadBalancer.K8sServices[svcns]
 	if ok && svc.IsHeadless {
-		err := d.policy.RegenerateToK8sServiceRules(svcns, *newSvcEP)
+		translator := k8s.NewK8sTranslator(svcns, *newSvcEP, false)
+		err := d.policy.TranslateRules(translator)
 		if err != nil {
 			log.Errorf("Unable to repopulate egress policies from ToService rules: %v", err)
 		}
@@ -560,7 +561,8 @@ func (d *Daemon) endpointDelFn(obj interface{}) {
 
 	svc, ok := d.loadBalancer.K8sServices[*svcns]
 	if ok && svc.IsHeadless {
-		err := d.policy.DeleteEndpointGeneratedEgressRulesWithLock(*svcns, endpoint)
+		translator := k8s.NewK8sTranslator(*svcns, endpoint, true)
+		err := d.policy.TranslateRules(translator)
 		if err != nil {
 			log.Errorf("Unable to depopulate egress policies from ToService rules: %v", err)
 		}
@@ -1065,7 +1067,8 @@ func (d *Daemon) addCiliumNetworkPolicy(obj interface{}) {
 
 	rules, err := rule.Parse()
 	if err == nil && len(rules) > 0 {
-		err = rules.GenerateEgressRulesFromEndpoints(
+		err = k8s.PreprocessRules(
+			rules,
 			d.loadBalancer.K8sEndpoints,
 			d.loadBalancer.K8sServices)
 		if err == nil {

pkg/k8s/rule_translate.go

@@ -0,0 +1,180 @@
+// Copyright 2016-2017 Authors of Cilium
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package k8s
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/cilium/cilium/common/types"
+	"github.com/cilium/cilium/pkg/policy"
+	"github.com/cilium/cilium/pkg/policy/api"
+)
+
+var _ policy.Translator = RuleTranslator{}
+
+// RuleTranslator implements pkg/policy.Translator interface
+// Translate populates/depopulates given rule with ToCIDR rules
+// Based on provided service/endpoint
+type RuleTranslator struct {
+	Service  types.K8sServiceNamespace
+	Endpoint types.K8sServiceEndpoint
+	Revert   bool
+}
+
+// Translate calls TranslateEgress on all r.Egress rules
+func (k RuleTranslator) Translate(r *api.Rule) error {
+	for egressIndex := range r.Egress {
+		err := k.TranslateEgress(&r.Egress[egressIndex])
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// TranslateEgress populates/depopulates egress rules with ToCIDR entries based
+// on toService entries
+func (k RuleTranslator) TranslateEgress(r *api.EgressRule) error {
+	err := k.depopulateEgress(r)
+	if err != nil {
+		return err
+	}
+	if !k.Revert {
+		err := k.populateEgress(r)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (k RuleTranslator) populateEgress(r *api.EgressRule) error {
+	for _, service := range r.ToServices {
+		// TODO: match services by labels
+		if service.K8sService == api.K8sServiceNamespace(k.Service) {
+			if err := generateToCidrFromEndpoint(r, k.Endpoint); err != nil {
+				return err
+			}
+			// TODO: generateToPortsFromEndpoint when ToPorts and ToCIDR are compatible
+		}
+	}
+	return nil
+}
+
+func (k RuleTranslator) depopulateEgress(r *api.EgressRule) error {
+	for _, service := range r.ToServices {
+		// TODO: match services by labels
+		if service.K8sService == api.K8sServiceNamespace(k.Service) {
+			if err := deleteToCidrFromEndpoint(r, k.Endpoint); err != nil {
+				return err
+			}
+			// TODO: generateToPortsFromEndpoint when ToPorts and ToCIDR are compatible
+		}
+	}
+	return nil
+}
+
+// generateToCidrFromEndpoint takes an egress rule and populates it with
+// ToCIDR rules based on provided endpoint object
+func generateToCidrFromEndpoint(
+	egress *api.EgressRule, endpoint types.K8sServiceEndpoint) error {
+
+	// This will generate one-address CIDRs consisting of endpoint backend ip
+	mask := net.CIDRMask(128, 128)
+	for ip := range endpoint.BEIPs {
+		epIP := net.ParseIP(ip)
+		if epIP == nil {
+			return fmt.Errorf("Unable to parse ip: %s", ip)
+		}
+
+		found := false
+		for _, c := range egress.ToCIDR {
+			_, cidr, err := net.ParseCIDR(string(c))
+			if err != nil {
+				return err
+			}
+			if cidr.Contains(epIP) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			cidr := net.IPNet{IP: epIP.Mask(mask), Mask: mask}
+			egress.ToCIDR = append(egress.ToCIDR, api.CIDR(cidr.String()))
+		}
+	}
+	return nil
+}
+
+// deleteToCidrFromEndpoint takes an egress rule and removes
+// ToCIDR rules matching endpoint
+func deleteToCidrFromEndpoint(
+	egress *api.EgressRule, endpoint types.K8sServiceEndpoint) error {
+
+	newToCIDR := make([]api.CIDR, 0, len(egress.ToCIDR))
+
+	for ip := range endpoint.BEIPs {
+		epIP := net.ParseIP(ip)
+		if epIP == nil {
+			return fmt.Errorf("Unable to parse ip: %s", ip)
+		}
+
+		for _, c := range egress.ToCIDR {
+			_, cidr, err := net.ParseCIDR(string(c))
+			if err != nil {
+				return err
+			}
+			if !cidr.Contains(epIP) {
+				//if endpoint is not in CIDR it's ok to retain it
+				newToCIDR = append(newToCIDR, c)
+			}
+		}
+	}
+
+	egress.ToCIDR = newToCIDR
+
+	return nil
+}
+
+// PreprocessRules translates rules that apply to headless services
+func PreprocessRules(
+	r api.Rules,
+	endpoints map[types.K8sServiceNamespace]*types.K8sServiceEndpoint,
+	services map[types.K8sServiceNamespace]*types.K8sServiceInfo) error {
+
+	for _, rule := range r {
+		for ns, ep := range endpoints {
+			svc, ok := services[ns]
+			if ok && svc.IsHeadless {
+				t := NewK8sTranslator(ns, *ep, false)
+				err := t.Translate(rule)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+// NewK8sTranslator returns RuleTranslator
+func NewK8sTranslator(
+	serviceInfo types.K8sServiceNamespace,
+	endpoint types.K8sServiceEndpoint,
+	revert bool) RuleTranslator {
+
+	return RuleTranslator{serviceInfo, endpoint, revert}
+}

pkg/k8s/rule_translate_test.go

@@ -0,0 +1,170 @@
+// Copyright 2016-2017 Authors of Cilium
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package k8s
+
+import (
+	"github.com/cilium/cilium/common/types"
+	"github.com/cilium/cilium/pkg/labels"
+	"github.com/cilium/cilium/pkg/policy"
+	"github.com/cilium/cilium/pkg/policy/api"
+
+	. "gopkg.in/check.v1"
+)
+
+func (s *K8sSuite) TestTranslator(c *C) {
+	repo := policy.NewPolicyRepository()
+
+	tag1 := labels.LabelArray{labels.ParseLabel("tag1")}
+	serviceInfo := types.K8sServiceNamespace{
+		ServiceName: "svc",
+		Namespace:   "default",
+	}
+
+	epIP := "10.1.1.1"
+
+	endpointInfo := types.K8sServiceEndpoint{
+		BEIPs: map[string]bool{
+			epIP: true,
+		},
+		Ports: map[types.FEPortName]*types.L4Addr{
+			"port": {
+				Protocol: types.TCP,
+				Port:     80,
+			},
+		},
+	}
+
+	rule1 := api.Rule{
+		EndpointSelector: api.NewESFromLabels(labels.ParseSelectLabel("bar")),
+		Egress: []api.EgressRule{{
+			ToServices: []api.Service{
+				{
+					K8sService: api.K8sServiceNamespace(serviceInfo),
+				},
+			}},
+		},
+		Labels: tag1,
+	}
+
+	translator := NewK8sTranslator(serviceInfo, endpointInfo, false)
+
+	_, err := repo.Add(rule1)
+	c.Assert(err, IsNil)
+
+	err = repo.TranslateRules(translator)
+	c.Assert(err, IsNil)
+
+	rule := repo.SearchRLocked(tag1)[0].Egress[0]
+
+	c.Assert(len(rule.ToCIDR), Equals, 1)
+	c.Assert(string(rule.ToCIDR[0]), Equals, epIP+"/32")
+
+	translator = NewK8sTranslator(serviceInfo, endpointInfo, true)
+	err = repo.TranslateRules(translator)
+
+	rule = repo.SearchRLocked(tag1)[0].Egress[0]
+
+	c.Assert(err, IsNil)
+	c.Assert(len(rule.ToCIDR), Equals, 0)
+}
+
+func (s *K8sSuite) TestGenerateToCIDRFromEndpoint(c *C) {
+	rule := &api.EgressRule{}
+
+	epIP := "10.1.1.1"
+
+	endpointInfo := types.K8sServiceEndpoint{
+		BEIPs: map[string]bool{
+			epIP: true,
+		},
+		Ports: map[types.FEPortName]*types.L4Addr{
+			"port": {
+				Protocol: types.TCP,
+				Port:     80,
+			},
+		},
+	}
+
+	err := generateToCidrFromEndpoint(rule, endpointInfo)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(rule.ToCIDR), Equals, 1)
+	c.Assert(string(rule.ToCIDR[0]), Equals, epIP+"/32")
+
+	// second run, to make sure there are no duplicates added
+	err = generateToCidrFromEndpoint(rule, endpointInfo)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(rule.ToCIDR), Equals, 1)
+	c.Assert(string(rule.ToCIDR[0]), Equals, epIP+"/32")
+
+	err = deleteToCidrFromEndpoint(rule, endpointInfo)
+	c.Assert(err, IsNil)
+	c.Assert(len(rule.ToCIDR), Equals, 0)
+}
+
+func (s *K8sSuite) TestPreprocessRules(c *C) {
+	tag1 := labels.LabelArray{labels.ParseLabel("tag1")}
+	serviceInfo := types.K8sServiceNamespace{
+		ServiceName: "svc",
+		Namespace:   "default",
+	}
+
+	epIP := "10.1.1.1"
+
+	endpointInfo := types.K8sServiceEndpoint{
+		BEIPs: map[string]bool{
+			epIP: true,
+		},
+		Ports: map[types.FEPortName]*types.L4Addr{
+			"port": {
+				Protocol: types.TCP,
+				Port:     80,
+			},
+		},
+	}
+
+	service := types.K8sServiceInfo{
+		IsHeadless: true,
+	}
+
+	rule1 := api.Rule{
+		EndpointSelector: api.NewESFromLabels(labels.ParseSelectLabel("bar")),
+		Egress: []api.EgressRule{{
+			ToServices: []api.Service{
+				{
+					K8sService: api.K8sServiceNamespace(serviceInfo),
+				},
+			}},
+		},
+		Labels: tag1,
+	}
+
+	endpoints := map[types.K8sServiceNamespace]*types.K8sServiceEndpoint{
+		serviceInfo: &endpointInfo,
+	}
+
+	services := map[types.K8sServiceNamespace]*types.K8sServiceInfo{
+		serviceInfo: &service,
+	}
+
+	rules := api.Rules{&rule1}
+
+	err := PreprocessRules(rules, endpoints, services)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(rule1.Egress[0].ToCIDR), Equals, 1)
+	c.Assert(string(rule1.Egress[0].ToCIDR[0]), Equals, epIP+"/32")
+}