switch to ownedby

Author: farodin91

coredns.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	log "github.com/sirupsen/logrus"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	etcdcv3 "go.etcd.io/etcd/client/v3"
 	"sigs.k8s.io/external-dns/pkg/tlsutils"
 
@@ -92,22 +93,19 @@ type Service struct {
 	// Etcd key where we found this service and ignored from json un-/marshaling
 	Key string `json:"-"`
 
-	// ManagedBy is used to prevent service to be added by different external-dns (only used by external-dns)
-	ManagedBy string `json:"managedby,omitempty"`
-
 	// OwnedBy is used to prevent service to be added by different external-dns (only used by external-dns)
 	OwnedBy string `json:"ownedby,omitempty"`
 }
 
 type etcdClient struct {
-	client               *etcdcv3.Client
-	managedBy            string
-	ignoreEmptyManagedBy bool
+	client        *etcdcv3.Client
+	ownerID       string
+	strictlyOwned bool
 }
 
 var _ coreDNSClient = etcdClient{}
 
-// GetServices return all Service records stored in etcd stored anywhere under the given key (recursively)
+// GetServices GetService return all Service records stored in etcd stored anywhere under the given key (recursively)
 func (c etcdClient) GetServices(ctx context.Context, prefix string) ([]*Service, error) {
 	ctx, cancel := context.WithTimeout(ctx, etcdTimeout)
 	defer cancel()
@@ -121,32 +119,26 @@ func (c etcdClient) GetServices(ctx context.Context, prefix string) ([]*Service,
 	var svcs []*Service
 	bx := make(map[Service]bool)
 	for _, n := range r.Kvs {
-		svc := new(Service)
-		if err := json.Unmarshal(n.Value, svc); err != nil {
-			return nil, fmt.Errorf("%s: %w", n.Key, err)
+		svc, err := c.unmarshalService(n)
+		if err != nil {
+			return nil, err
+		}
+		if c.strictlyOwned && svc.OwnedBy != c.ownerID {
+			continue
 		}
 		b := Service{
-			Host:      svc.Host,
-			Port:      svc.Port,
-			Priority:  svc.Priority,
-			Weight:    svc.Weight,
-			Text:      svc.Text,
-			Key:       string(n.Key),
-			ManagedBy: svc.ManagedBy,
-			OwnedBy:   svc.OwnedBy,
+			Host:     svc.Host,
+			Port:     svc.Port,
+			Priority: svc.Priority,
+			Weight:   svc.Weight,
+			Text:     svc.Text,
+			Key:      string(n.Key),
 		}
 		if _, ok := bx[b]; ok {
 			// skip the service if already added to service list.
 			// the same service might be found in multiple etcd nodes.
 			continue
 		}
-		if c.managedBy != "" {
-			if c.ignoreEmptyManagedBy && b.ManagedBy != c.managedBy {
-				continue
-			} else if !c.ignoreEmptyManagedBy && b.ManagedBy != "" && b.ManagedBy != c.managedBy {
-				continue
-			}
-		}
 		bx[b] = true
 
 		svc.Key = string(n.Key)
@@ -155,7 +147,6 @@ func (c etcdClient) GetServices(ctx context.Context, prefix string) ([]*Service,
 		}
 		svcs = append(svcs, svc)
 	}
-
 	return svcs, nil
 }
 
@@ -164,14 +155,23 @@ func (c etcdClient) SaveService(ctx context.Context, service *Service) error {
 	ctx, cancel := context.WithTimeout(ctx, etcdTimeout)
 	defer cancel()
 
-	if c.managedBy != "" {
-		service.ManagedBy = c.managedBy
-		service.OwnedBy = c.managedBy
-	}
-	if ownedBy, err := c.IsOwnedBy(ctx, service.Key); err != nil {
-		return err
-	} else if !ownedBy {
-		return fmt.Errorf("key %q is not owned by this service", service.Key)
+	// check only for empty OwnedBy
+	if c.strictlyOwned && service.OwnedBy != c.ownerID {
+		r, err := c.client.Get(ctx, service.Key)
+		if err != nil {
+			return fmt.Errorf("etcd get %q: %w", service.Key, err)
+		}
+		// Key missing -> treat as owned (safe to create)
+		if r != nil && len(r.Kvs) != 0 {
+			svc, err := c.unmarshalService(r.Kvs[0])
+			if err != nil {
+				return fmt.Errorf("failed to unmarshal value for key %q: %w", service.Key, err)
+			}
+			if svc.OwnedBy != c.ownerID {
+				return fmt.Errorf("key %q is not owned by this provider", service.Key)
+			}
+		}
+		service.OwnedBy = c.ownerID
 	}
 
 	value, err := json.Marshal(&service)
@@ -185,54 +185,43 @@ func (c etcdClient) SaveService(ctx context.Context, service *Service) error {
 	return nil
 }
 
-func (c etcdClient) IsOwnedBy(ctx context.Context, key string) (bool, error) {
+// DeleteService deletes service record from etcd
+func (c etcdClient) DeleteService(ctx context.Context, key string) error {
 	ctx, cancel := context.WithTimeout(ctx, etcdTimeout)
 	defer cancel()
 
-	if c.managedBy == "" {
-		return true, nil
-	}
-
-	r, err := c.client.Get(ctx, key)
-	if err != nil {
-		return false, err
-	}
-	if r == nil {
-		return true, nil
-	} else if len(r.Kvs) > 1 {
-		return false, fmt.Errorf("found multiple keys with the same key this service")
-	} else if len(r.Kvs) == 0 {
-		return true, nil
-	}
-	for _, n := range r.Kvs {
-		svc := new(Service)
-		if err := json.Unmarshal(n.Value, svc); err != nil {
-			return false, fmt.Errorf("%s: %w", n.Key, err)
+	if c.strictlyOwned {
+		rs, err := c.client.Get(ctx, key, etcdcv3.WithPrefix())
+		if err != nil {
+			return err
 		}
+		for _, r := range rs.Kvs {
+			svc, err := c.unmarshalService(r)
+			if err != nil {
+				return err
+			}
+			if svc.OwnedBy != c.ownerID {
+				continue
+			}
 
-		if !c.ignoreEmptyManagedBy && svc.ManagedBy == "" {
-			return true, nil
-		}
-		if svc.ManagedBy == c.managedBy {
-			return true, nil
+			_, err = c.client.Delete(ctx, string(r.Key))
+			if err != nil {
+				return err
+			}
 		}
+		return err
+	} else {
+		_, err := c.client.Delete(ctx, key, etcdcv3.WithPrefix())
+		return err
 	}
-	return false, nil
 }
 
-// DeleteService deletes service record from etcd
-func (c etcdClient) DeleteService(ctx context.Context, key string) error {
-	ctx, cancel := context.WithTimeout(ctx, etcdTimeout)
-	defer cancel()
-
-	if owned, err := c.IsOwnedBy(ctx, key); err != nil {
-		return err
-	} else if !owned {
-		return fmt.Errorf("key %q is not owned by this service", key)
+func (c etcdClient) unmarshalService(n *mvccpb.KeyValue) (*Service, error) {
+	svc := new(Service)
+	if err := json.Unmarshal(n.Value, svc); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal %q: %w", n.Key, err)
 	}
-
-	_, err := c.client.Delete(ctx, key, etcdcv3.WithPrefix())
-	return err
+	return svc, nil
 }
 
 // builds etcd client config depending on connection scheme and TLS parameters
@@ -265,7 +254,7 @@ func getETCDConfig() (*etcdcv3.Config, error) {
 }
 
 // the newETCDClient is an etcd client constructor
-func newETCDClient(managedBy string, ignoreEmptyManagedBy bool) (coreDNSClient, error) {
+func newETCDClient(ownerID string, strictlyOwned bool) (coreDNSClient, error) {
 	cfg, err := getETCDConfig()
 	if err != nil {
 		return nil, err
@@ -274,12 +263,12 @@ func newETCDClient(managedBy string, ignoreEmptyManagedBy bool) (coreDNSClient,
 	if err != nil {
 		return nil, err
 	}
-	return etcdClient{c, managedBy, ignoreEmptyManagedBy}, nil
+	return etcdClient{c, ownerID, strictlyOwned}, nil
 }
 
 // NewCoreDNSProvider is a CoreDNS provider constructor
-func NewCoreDNSProvider(config CoreDNSConfig, managedBy string, ignoreEmptyManagedBy, dryRun bool) (provider.Provider, error) {
-	client, err := newETCDClient(managedBy, ignoreEmptyManagedBy)
+func NewCoreDNSProvider(config CoreDNSConfig, ownerID string, strictlyOwned, dryRun bool) (provider.Provider, error) {
+	client, err := newETCDClient(ownerID, strictlyOwned)
 	if err != nil {
 		return nil, err
 	}