<|start_filename|>util/cmd/csv.go<|end_filename|>
package cmd

import (
	"fmt"
	"strings"
)

// CommaSeparatedValues represents a slice of strings that were originally separated by ','.
type CommaSeparatedValues []string

func (c *CommaSeparatedValues) String() string {
	return fmt.Sprint(*c)
}

// Set binds a comma separated command line flag value to a KeyValue.
func (c *CommaSeparatedValues) Set(value string) error {
	if value != "" {
		*c = strings.Split(value, ",")
	}
	return nil
}


<|start_filename|>feed-ingress/cmd/gorb.go<|end_filename|>
package cmd

import (
	"fmt"
	"strconv"
	"strings"

	"github.com/sky-uk/feed/k8s"

	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/gorb"
	"github.com/spf13/cobra"
)

var (
	gorbIngressInstanceIP          string
	gorbEndpoint                   string
	gorbServicesDefinition         string
	gorbBackendMethod              string
	gorbBackendWeight              int
	gorbVipLoadbalancer            string
	gorbManageLoopback             bool
	gorbBackendHealthcheckInterval string
	gorbBackendHealthcheckType     string
	gorbInterfaceProcFsPath        string
)

const (
	defaultGorbIngressInstanceIP          = "127.0.0.1"
	defaultGorbEndpoint                   = "http://127.0.0.1:80"
	defaultGorbBackendMethod              = "dr"
	defaultGorbBackendWeight              = 1000
	defaultGorbServicesDefinition         = "http-proxy:80,https-proxy:443"
	defaultGorbVipLoadbalancer            = "127.0.0.1"
	defaultGorbManageLoopback             = true
	defaultGorbInterfaceProcFsPath        = "/host-ipv4-proc/"
	defaultGorbBackendHealthcheckInterval = "1s"
	defaultGorbBackendHealthcheckType     = "http"
)

var gorbCmd = &cobra.Command{
	Use:   "gorb",
	Short: "Attach to Gorb Load Balancers (deprecated; use Merlin instead)",
	Long:  `Configure IPVS via Gorb. https://github.com/sky-uk/gorb`,
	Run: func(cmd *cobra.Command, args []string) {
		runCmd(appendGorbIngressUpdaters)
	},
}

func init() {
	rootCmd.AddCommand(gorbCmd)

	gorbCmd.Flags().StringVar(&gorbEndpoint, "gorb-endpoint", defaultGorbEndpoint, "Define the endpoint to talk to gorb for registration.")
	gorbCmd.Flags().StringVar(&gorbIngressInstanceIP, "gorb-ingress-instance-ip", defaultGorbIngressInstanceIP,
		"Define the ingress instance ip, the ip of the node where feed-ingress is running.")
	gorbCmd.Flags().StringVar(&gorbServicesDefinition, "gorb-services-definition", defaultGorbServicesDefinition,
		"Comma separated list of Service Definition (e.g. 'http-proxy:80,https-proxy:443') to register via Gorb")
	gorbCmd.Flags().StringVar(&gorbBackendMethod, "gorb-backend-method", defaultGorbBackendMethod,
		"Define the backend method (e.g. nat, dr, tunnel) to register via Gorb ")
	gorbCmd.Flags().IntVar(&gorbBackendWeight, "gorb-backend-weight", defaultGorbBackendWeight,
		"Define the backend weight to register via Gorb")
	gorbCmd.Flags().StringVar(&gorbVipLoadbalancer, "gorb-vip-loadbalancer", defaultGorbVipLoadbalancer,
		"Define the vip loadbalancer to set the loopback. Only necessary when Direct Return is enabled.")
	gorbCmd.Flags().BoolVar(&gorbManageLoopback, "gorb-management-loopback", defaultGorbManageLoopback,
		"Enable loopback creation. Only necessary when Direct Return is enabled")
	gorbCmd.Flags().StringVar(&gorbInterfaceProcFsPath, "gorb-interface-proc-fs-path", defaultGorbInterfaceProcFsPath,
		"Path to the interface proc file system. Only necessary when Direct Return is enabled")
	gorbCmd.Flags().StringVar(&gorbBackendHealthcheckInterval, "gorb-backend-healthcheck-interval", defaultGorbBackendHealthcheckInterval,
		"Define the gorb healthcheck interval for the backend")
	gorbCmd.Flags().StringVar(&gorbBackendHealthcheckType, "gorb-backend-healthcheck-type", defaultGorbBackendHealthcheckType,
		"Define the gorb healthcheck type for the backend. Must be either 'tcp', 'http' or 'none'")
}

func appendGorbIngressUpdaters(kubernetesClient k8s.Client, updaters []controller.Updater) ([]controller.Updater, error) {
	virtualServices, err := toVirtualServices(gorbServicesDefinition)
	if err != nil {
		return nil, fmt.Errorf("invalid gorb services definition. Must be a comma separated list - e.g. 'http-proxy:80,https-proxy:443', but was %s", gorbServicesDefinition)
	}

	if gorbBackendHealthcheckType != "tcp" && gorbBackendHealthcheckType != "http" && gorbBackendHealthcheckType != "none" {
		return nil, fmt.Errorf("invalid gorb backend healthcheck type. Must be either 'tcp', 'http' or 'none', but was %s", gorbBackendHealthcheckType)
	}

	config := gorb.Config{
		ServerBaseURL:              gorbEndpoint,
		InstanceIP:                 gorbIngressInstanceIP,
		DrainDelay:                 drainDelay,
		ServicesDefinition:         virtualServices,
		BackendMethod:              gorbBackendMethod,
		BackendWeight:              gorbBackendWeight,
		VipLoadbalancer:            gorbVipLoadbalancer,
		ManageLoopback:             gorbManageLoopback,
		BackendHealthcheckInterval: gorbBackendHealthcheckInterval,
		BackendHealthcheckType:     gorbBackendHealthcheckType,
		InterfaceProcFsPath:        gorbInterfaceProcFsPath,
	}

	gorbUpdater, err := gorb.New(&config)
	if err != nil {
		return nil, err
	}
	return append(updaters, gorbUpdater), nil
}

func toVirtualServices(servicesCsv string) ([]gorb.VirtualService, error) {
	virtualServices := make([]gorb.VirtualService, 0)
	servicesDefinitionArr := strings.Split(servicesCsv, ",")
	for _, service := range servicesDefinitionArr {
		servicesArr := strings.Split(service, ":")
		if len(servicesArr) != 2 {
			return nil, fmt.Errorf("unable to convert %s to servicename:port combination", servicesArr)
		}
		port, err := strconv.Atoi(servicesArr[1])
		if err != nil {
			return nil, fmt.Errorf("unable to convert port %s to int", servicesArr[1])
		}
		virtualServices = append(virtualServices, gorb.VirtualService{Name: servicesArr[0], Port: port})
	}
	return virtualServices, nil
}


<|start_filename|>feed-ingress/cmd/common_test.go<|end_filename|>
package cmd

import (
	"os"
	"os/exec"
	"testing"

	"github.com/stretchr/testify/assert"

	"github.com/sky-uk/feed/nginx"
)

func TestCreatePortsConfigWithoutPorts(t *testing.T) {
	if os.Getenv("BE_CRASHER") == "1" {
		createPortsConfig(unset, unset)
		return
	}
	cmd := exec.Command(os.Args[0], "-test.run=TestCreatePortsConfigWithoutPorts")
	cmd.Env = append(os.Environ(), "BE_CRASHER=1")
	err := cmd.Run()
	if e, ok := err.(*exec.ExitError); ok && !e.Success() {
		return
	}
	t.Fatalf("process ran with err %v, want exit status 1", err)
}

func TestCreatePortsConfigWithOnePort(t *testing.T) {
	ports := createPortsConfig(1, unset)
	expectedPorts := []nginx.Port{nginx.Port{Name: "http", Port: 1}}
	portsHTTPS := createPortsConfig(unset, 2)
	expectedPortsHTTPS := []nginx.Port{nginx.Port{Name: "https", Port: 2}}

	assert.Equal(t, expectedPorts, ports, "they should be equal")
	assert.Equal(t, expectedPortsHTTPS, portsHTTPS, "they should be equal")
}

func TestCreatePortsConfigWithPorts(t *testing.T) {
	ports := createPortsConfig(1, 2)
	expectedPorts := []nginx.Port{nginx.Port{Name: "http", Port: 1}, nginx.Port{Name: "https", Port: 2}}

	assert.Equal(t, expectedPorts, ports, "they should be equal")
}


<|start_filename|>merlin/merlin.go<|end_filename|>
package merlin

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/golang/protobuf/proto"
	"github.com/golang/protobuf/ptypes"
	"github.com/golang/protobuf/ptypes/wrappers"
	log "github.com/sirupsen/logrus"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/merlin/types"
	"google.golang.org/grpc"
	"google.golang.org/grpc/balancer/roundrobin"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// Config for merlin updater.
type Config struct {
	Endpoint            string
	Timeout             time.Duration
	ServiceID           string
	HTTPSServiceID      string
	InstanceIP          string
	InstancePort        uint16
	InstanceHTTPSPort   uint16
	ForwardMethod       string
	HealthPort          uint16
	HealthPath          string
	HealthUpThreshold   uint32
	HealthDownThreshold uint32
	HealthPeriod        time.Duration
	HealthTimeout       time.Duration
	DrainDelay          time.Duration
	VIP                 string
	VIPInterface        string
}

type closeable interface {
	Close() error
}

type updater struct {
	Config
	clientFactory func(*Config) (types.MerlinClient, closeable, error)
	nl            netlinkWrapper
}

// New merlin updater.
func New(conf Config) (controller.Updater, error) {
	u := &updater{
		Config:        conf,
		clientFactory: createRealClient,
		nl:            &netlinkWrapperImpl{},
	}
	return u, nil
}

func createRealClient(conf *Config) (types.MerlinClient, closeable, error) {
	c, err := grpc.Dial(conf.Endpoint, grpc.WithInsecure(), grpc.WithBalancerName(roundrobin.Name))
	if err != nil {
		return nil, nil, fmt.Errorf("unable to create merlin grpc client: %v", err)
	}
	client := types.NewMerlinClient(c)
	return client, c, nil
}

func (u *updater) Start() error {
	if err := u.addVIP(); err != nil {
		return err
	}
	if err := u.registerWithMerlin(); err != nil {
		if err := u.removeVIP(); err != nil {
			log.Warnf("Unable to remove VIP: %v", err)
		}
		return err
	}
	return nil
}

func (u *updater) createBaseRealServer() *types.RealServer {
	return &types.RealServer{
		ServiceID: u.ServiceID,
		Key: &types.RealServer_Key{
			Ip:   u.InstanceIP,
			Port: uint32(u.InstancePort),
		},
	}
}

func (u *updater) createHTTPSFrom(orig *types.RealServer) *types.RealServer {
	server := proto.Clone(orig).(*types.RealServer)
	server.ServiceID = u.HTTPSServiceID
	server.Key.Port = uint32(u.InstanceHTTPSPort)
	return server
}

func (u *updater) registerWithMerlin() error {
	// create merlin server values
	forward, ok := types.ForwardMethod_value[strings.ToUpper(u.ForwardMethod)]
	if !ok {
		return fmt.Errorf("unrecognized forward method: %s", u.ForwardMethod)
	}
	server := u.createBaseRealServer()
	server.Config = &types.RealServer_Config{
		Weight:  &wrappers.UInt32Value{Value: 1},
		Forward: types.ForwardMethod(forward),
	}
	server.HealthCheck = &types.RealServer_HealthCheck{
		Endpoint:      &wrappers.StringValue{Value: fmt.Sprintf("http://:%d/%s", u.HealthPort, u.HealthPath)},
		UpThreshold:   u.HealthUpThreshold,
		DownThreshold: u.HealthDownThreshold,
		Period:        ptypes.DurationProto(u.HealthPeriod),
		Timeout:       ptypes.DurationProto(u.HealthTimeout),
	}

	// create gRPC client
	client, conn, err := u.clientFactory(&u.Config)
	if err != nil {
		return err
	}
	defer conn.Close()

	// register with merlin
	if err := u.registerServer(client, server, "http"); err != nil {
		return err
	}
	httpsServer := u.createHTTPSFrom(server)
	return u.registerServer(client, httpsServer, "https")
}

func (u *updater) registerServer(client types.MerlinClient, server *types.RealServer, detail string) error {
	if server.ServiceID == "" || server.Key.Port == 0 {
		log.Infof("Skipping merlin registration for %s", detail)
		return nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), u.Timeout)
	defer cancel()

	_, err := client.CreateServer(ctx, server)
	if status.Code(err) == codes.AlreadyExists {
		if _, err := client.UpdateServer(ctx, server); err != nil {
			return fmt.Errorf("unable to register to %s in merlin: %v", server.ServiceID, err)
		}
	} else if err != nil {
		return fmt.Errorf("unable to register to %s in merlin: %v", server.ServiceID, err)
	}

	log.Infof("Successfully registered to %s in merlin", server.ServiceID)
	return nil
}

func (u *updater) Stop() error {
	u.deregisterWithMerlin()
	return u.removeVIP()
}

func (u *updater) deregisterWithMerlin() {
	// create gRPC client
	client, conn, err := u.clientFactory(&u.Config)
	if err != nil {
		log.Warnf("Unable to create client connection to merlin: %v", err)
	}
	defer conn.Close()

	// create server keys
	server := u.createBaseRealServer()
	httpsServer := u.createHTTPSFrom(server)

	// drain
	u.updateServerForDraining(client, server)
	u.updateServerForDraining(client, httpsServer)
	time.Sleep(u.DrainDelay)

	// deregister
	u.deregisterServer(client, server)
	u.deregisterServer(client, httpsServer)
}

func (u *updater) updateServerForDraining(client types.MerlinClient, orig *types.RealServer) {
	if orig.ServiceID == "" || orig.Key.Port == 0 {
		return
	}
	server := proto.Clone(orig).(*types.RealServer)
	server.Config = &types.RealServer_Config{Weight: &wrappers.UInt32Value{Value: 0}}
	ctx, cancel := context.WithTimeout(context.Background(), u.Timeout)
	defer cancel()
	if _, err := client.UpdateServer(ctx, server); err != nil {
		log.Warnf("Draining failed for %s, unable to set weight to 0: %v", server.ServiceID, err)
	} else {
		log.Infof("Started draining for %s, server weight set to 0", server.ServiceID)
	}
}

func (u *updater) deregisterServer(client types.MerlinClient, server *types.RealServer) {
	if server.ServiceID == "" || server.Key.Port == 0 {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), u.Timeout)
	defer cancel()
	if _, err := client.DeleteServer(ctx, server); err != nil {
		log.Errorf("Unable to deregister from %s in merlin, please remove manually: %v", server.ServiceID, err)
	} else {
		log.Infof("Successfully deregistered from %s in merlin", server.ServiceID)
	}
}

func (u *updater) addVIP() error {
	if u.VIPInterface == "" || u.VIP == "" {
		return nil
	}
	return u.nl.addVIP(u.VIPInterface, u.VIP)
}

func (u *updater) removeVIP() error {
	if u.VIPInterface == "" || u.VIP == "" {
		return nil
	}
	return u.nl.removeVIP(u.VIPInterface, u.VIP)
}

func (u *updater) Update(controller.IngressEntries) error {
	return nil
}

func (u *updater) Health() error {
	return nil
}

func (u *updater) String() string {
	return "<NAME>"
}


<|start_filename|>alb/alb_metrics.go<|end_filename|>
package alb

import (
	"sync"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/sky-uk/feed/util/metrics"
)

var once sync.Once
var attachedFrontendGauge prometheus.Gauge

func initMetrics() {
	once.Do(func() {
		attachedFrontendGauge = metrics.RegisterNewDefaultGauge(metrics.PrometheusIngressSubsystem,
			"alb_frontends_attached", "The total number of ALB frontends attached to")
	})
}


<|start_filename|>nlb/nlbstatus/status.go<|end_filename|>
/*
Package nlbstatus provides an updater for an NLB frontend to update ingress statuses.
*/
package nlbstatus

import (
	"fmt"

	"github.com/sky-uk/feed/nlb"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/elbv2"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/k8s"
	k8sStatus "github.com/sky-uk/feed/k8s/status"
	v1 "k8s.io/api/core/v1"
)

// Config for creating a new NLB status updater.
type Config struct {
	Region              string
	FrontendTagValue    string
	IngressNameTagValue string
	KubernetesClient    k8s.Client
}

// New creates a new NLB frontend status updater.
func New(conf Config) (controller.Updater, error) {
	awsSession, err := session.NewSession(&aws.Config{Region: &conf.Region})
	if err != nil {
		return nil, fmt.Errorf("unable to create NLB status updater: %v", err)
	}

	return &status{
		awsElb:              elbv2.New(awsSession),
		frontendTagValue:    conf.FrontendTagValue,
		ingressNameTagValue: conf.IngressNameTagValue,
		loadBalancers:       make(map[string]v1.LoadBalancerStatus),
		kubernetesClient:    conf.KubernetesClient,
	}, nil
}

type status struct {
	awsElb              nlb.ELBV2
	frontendTagValue    string
	ingressNameTagValue string
	loadBalancers       map[string]v1.LoadBalancerStatus
	kubernetesClient    k8s.Client
}

// Start discovers the NLBs and generates loadBalancer statuses.
func (s *status) Start() error {
	clusterFrontEnds, err := nlb.FindFrontEndLoadBalancersWithIngressClassName(s.awsElb, s.frontendTagValue, s.ingressNameTagValue)
	if err != nil {
		return err
	}

	for lbLabel, clusterFrontEnd := range clusterFrontEnds {
		s.loadBalancers[lbLabel] = k8sStatus.GenerateLoadBalancerStatus([]string{clusterFrontEnd.DNSName})
	}
	return nil
}

func (s *status) Stop() error {
	return nil
}

func (s *status) Health() error {
	return nil
}

func (s *status) Update(ingresses controller.IngressEntries) error {
	return k8sStatus.Update(ingresses, s.loadBalancers, s.kubernetesClient)
}


<|start_filename|>alb/alb.go<|end_filename|>
/*
Package alb provides an updater for an ALB frontend to attach NGINX to.
*/
package alb

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/ec2metadata"
	"github.com/aws/aws-sdk-go/aws/session"
	aws_alb "github.com/aws/aws-sdk-go/service/elbv2"
	log "github.com/sirupsen/logrus"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/util"
)

// New creates a controller.Updater for attaching to ALB target groups on first update.
func New(region string, targetGroupNames []string, targetGroupDeregistrationDelay time.Duration) (controller.Updater, error) {
	if len(targetGroupNames) == 0 {
		return nil, errors.New("unable to create ALB updater: missing target group names")
	}
	initMetrics()
	log.Infof("ALB frontend region: %s target groups: %v", region, targetGroupNames)
	awsSession, err := session.NewSession(&aws.Config{Region: &region})

	if err != nil {
		return nil, fmt.Errorf("unable to create ALB updater: %v", err)
	}

	return &alb{
		metadata:                       ec2metadata.New(awsSession),
		awsALB:                         aws_alb.New(awsSession),
		targetGroupNames:               targetGroupNames,
		targetGroupDeregistrationDelay: targetGroupDeregistrationDelay,
		region:                         region,
		initialised:                    initialised{},
	}, nil
}

type alb struct {
	awsALB                         ALB
	metadata                       EC2Metadata
	targetGroupNames               []string
	targetGroupDeregistrationDelay time.Duration
	region                         string
	instanceID                     string
	albARNs                        []*string
	registeredFrontends            util.SafeInt
	initialised                    initialised
	readyForHealthCheck            util.SafeBool
}

type initialised struct {
	sync.Mutex
	done bool
}

// ALB interface to allow mocking of real calls to AWS as well as cutting down the methods from the real
// interface to only the ones we use
type ALB interface {
	DescribeTargetGroups(input *aws_alb.DescribeTargetGroupsInput) (*aws_alb.DescribeTargetGroupsOutput, error)
	RegisterTargets(input *aws_alb.RegisterTargetsInput) (*aws_alb.RegisterTargetsOutput, error)
	DeregisterTargets(input *aws_alb.DeregisterTargetsInput) (*aws_alb.DeregisterTargetsOutput, error)
}

// EC2Metadata interface to allow mocking of the real calls to AWS
type EC2Metadata interface {
	Available() bool
	Region() (string, error)
	GetInstanceIdentityDocument() (ec2metadata.EC2InstanceIdentityDocument, error)
}

func (a *alb) Start() error {
	return nil
}

func (a *alb) Update(controller.IngressEntries) error {
	a.initialised.Lock()
	defer a.initialised.Unlock()
	defer func() { a.readyForHealthCheck.Set(true) }()

	if !a.initialised.done {
		log.Infof("Attaching to ALB target groups: %v", a.targetGroupNames)
		if err := a.attachToFrontEnds(); err != nil {
			return err
		}
		a.initialised.done = true
	}
	return nil
}

// Stop removes this instance from all the front end ALBs
func (a *alb) Stop() error {
	for _, arn := range a.albARNs {
		log.Infof("Deregistering instance %s with ALB target group %s", a.instanceID, *arn)

		_, err := a.awsALB.DeregisterTargets(&aws_alb.DeregisterTargetsInput{
			TargetGroupArn: arn,
			Targets: []*aws_alb.TargetDescription{
				{Id: aws.String(a.instanceID)},
			},
		})

		if err != nil {
			log.Warnf("Unable to deregister instance %s with ALB target group %s: %v", a.instanceID, *arn, err)
		}
	}

	log.Infof("Waiting %v to finish ALB target group deregistration", a.targetGroupDeregistrationDelay)
	time.Sleep(a.targetGroupDeregistrationDelay)

	return nil
}

// Health returns nil if attached to all frontends.
func (a *alb) Health() error {
	if !a.readyForHealthCheck.Get() || len(a.targetGroupNames) == a.registeredFrontends.Get() {
		return nil
	}
	return fmt.Errorf("have not attached to all frontends %v yet", a.targetGroupNames)
}

func (a *alb) String() string {
	return "ELB frontend"
}

func (a *alb) attachToFrontEnds() error {
	if len(a.targetGroupNames) == 0 {
		return nil
	}

	instanceDoc, err := a.metadata.GetInstanceIdentityDocument()
	if err != nil {
		return fmt.Errorf("unable to query ec2 metadata service for InstanceId: %v", err)
	}
	instanceID := instanceDoc.InstanceID
	a.instanceID = instanceID

	arns, err := a.findTargetGroupARNs(a.targetGroupNames)
	if err != nil {
		return err
	}
	log.Infof("Found %d front ends", len(arns))
	a.albARNs = arns

	registered := 0
	for _, arn := range arns {
		log.Infof("Registering instance %s with alb %s", instanceID, *arn)

		_, err = a.awsALB.RegisterTargets(&aws_alb.RegisterTargetsInput{
			TargetGroupArn: arn,
			Targets: []*aws_alb.TargetDescription{
				{Id: aws.String(instanceID)},
			},
		})

		if err != nil {
			return fmt.Errorf("unable to register instance %s with ALB target group %s: %v", instanceID, *arn, err)
		}
		registered++
	}

	attachedFrontendGauge.Set(float64(registered))
	a.registeredFrontends.Set(registered)

	if len(a.targetGroupNames) != registered {
		return fmt.Errorf("only attached to %d ALBs, expected %d", registered, len(a.targetGroupNames))
	}

	return nil
}

func (a *alb) findTargetGroupARNs(names []string) ([]*string, error) {
	req := &aws_alb.DescribeTargetGroupsInput{Names: aws.StringSlice(names)}
	var arns []*string

	for {
		resp, err := a.awsALB.DescribeTargetGroups(req)
		if err != nil {
			return nil, err
		}

		for _, targetGroup := range resp.TargetGroups {
			arns = append(arns, targetGroup.TargetGroupArn)
		}

		if resp.NextMarker == nil {
			break
		}

		req.Marker = resp.NextMarker
	}

	return arns, nil
}


<|start_filename|>feed-ingress/cmd/merlin.go<|end_filename|>
package cmd

import (
	"time"

	"github.com/sky-uk/feed/merlin"

	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/k8s"

	"github.com/sky-uk/feed/merlin/status"
	"github.com/spf13/cobra"
)

var (
	merlinEndpoint               string
	merlinRequestTimeout         time.Duration
	merlinServiceID              string
	merlinHTTPSServiceID         string
	merlinInstanceIP             string
	merlinForwardMethod          string
	merlinDrainDelay             time.Duration
	merlinHealthUpThreshold      uint
	merlinHealthDownThreshold    uint
	merlinHealthPeriod           time.Duration
	merlinHealthTimeout          time.Duration
	merlinVIP                    string
	merlinVIPInterface           string
	merlinInternalHostname       string
	merlinInternetFacingHostname string
)

const (
	defaultMerlinForwardMethod       = "route"
	defaultMerlinHealthUpThreshold   = 3
	defaultMerlinHealthDownThreshold = 2
	defaultMerlinHealthPeriod        = 10 * time.Second
	defaultMerlinHealthTimeout       = time.Second
	defaultMerlinVIPInterface        = "lo"
)

var merlinCmd = &cobra.Command{
	Use:   "merlin",
	Short: "Attach to Merlin Load Balancers",
	Long: `Merlin is a distributed load balancer based on IPVS, with a gRPC based API.
Feed Ingress supports attaching to merlin as a frontend. https://github.com/sky-uk/merlin`,
	Run: func(cmd *cobra.Command, args []string) {
		runCmd(appendMerlinIngressUpdaters)
	},
}

func init() {
	rootCmd.AddCommand(merlinCmd)

	merlinCmd.Flags().StringVar(&merlinEndpoint, "merlin-endpoint", "",
		"Merlin gRPC endpoint to connect to. Expected format is scheme://authority/endpoint_name (see "+
			"https://github.com/grpc/grpc/blob/master/doc/naming.md). Will load balance between all available servers.")
	merlinCmd.Flags().DurationVar(&merlinRequestTimeout, "merlin-request-timeout", time.Second*10,
		"Timeout for any requests to merlin.")
	merlinCmd.Flags().StringVar(&merlinServiceID, "merlin-service-id", "", "Merlin http virtual service ID to attach to.")
	merlinCmd.Flags().StringVar(&merlinHTTPSServiceID, "merlin-https-service-id", "", "Merlin https virtual service ID to attach to.")
	merlinCmd.Flags().StringVar(&merlinInstanceIP, "merlin-instance-ip", "", "Ingress IP to register with merlin")
	merlinCmd.Flags().StringVar(&merlinForwardMethod, "merlin-forward-method", defaultMerlinForwardMethod, "IPVS forwarding method,"+
		" must be one of route, tunnel, or masq.")
	merlinCmd.Flags().DurationVar(&merlinDrainDelay, "merlin-drain-delay", defaultDrainDelay, "Delay to wait after for connections"+
		" to bleed off when deregistering from merlin. Real server weight is set to 0 during this delay.")
	merlinCmd.Flags().UintVar(&merlinHealthUpThreshold, "merlin-health-up-threshold", defaultMerlinHealthUpThreshold,
		"Number of checks before merlin will consider this instance healthy.")
	merlinCmd.Flags().UintVar(&merlinHealthDownThreshold, "merlin-health-down-threshold", defaultMerlinHealthDownThreshold,
		"Number of checks before merlin will consider this instance unhealthy.")
	merlinCmd.Flags().DurationVar(&merlinHealthPeriod, "merlin-health-period", defaultMerlinHealthPeriod,
		"The time between health checks.")
	merlinCmd.Flags().DurationVar(&merlinHealthTimeout, "merlin-health-timeout", defaultMerlinHealthTimeout,
		"The timeout for health checks.")
	merlinCmd.Flags().StringVar(&merlinVIP, "merlin-vip", "", "VIP to assign to loopback to support direct route and tunnel.")
	merlinCmd.Flags().StringVar(&merlinVIPInterface, "merlin-vip-interface", defaultMerlinVIPInterface,
		"VIP interface to assign the VIP.")
	merlinCmd.Flags().StringVar(&merlinInternalHostname, "merlin-internal-hostname", "",
		"Hostname of the internal facing load-balancer.")
	merlinCmd.Flags().StringVar(&merlinInternetFacingHostname, "merlin-internet-facing-hostname", "",
		"Hostname of the internet facing load-balancer")
}

func appendMerlinIngressUpdaters(kubernetesClient k8s.Client, updaters []controller.Updater) ([]controller.Updater, error) {
	config := merlin.Config{
		Endpoint:          merlinEndpoint,
		Timeout:           merlinRequestTimeout,
		ServiceID:         merlinServiceID,
		HTTPSServiceID:    merlinHTTPSServiceID,
		InstanceIP:        merlinInstanceIP,
		InstancePort:      uint16(ingressPort),
		InstanceHTTPSPort: uint16(ingressHTTPSPort),
		ForwardMethod:     merlinForwardMethod,
		DrainDelay:        merlinDrainDelay,
		HealthPort:        uint16(ingressHealthPort),
		// This value is hardcoded into the nginx template.
		HealthPath:          "health",
		HealthUpThreshold:   uint32(merlinHealthUpThreshold),
		HealthDownThreshold: uint32(merlinHealthDownThreshold),
		HealthPeriod:        merlinHealthPeriod,
		HealthTimeout:       merlinHealthTimeout,
		VIP:                 merlinVIP,
		VIPInterface:        merlinVIPInterface,
	}
	merlinUpdater, err := merlin.New(config)
	if err != nil {
		return nil, err
	}
	updaters = append(updaters, merlinUpdater)

	if merlinInternalHostname != "" || merlinInternetFacingHostname != "" {
		statusConfig := status.Config{
			InternalHostname:       merlinInternalHostname,
			InternetFacingHostname: merlinInternetFacingHostname,
			KubernetesClient:       kubernetesClient,
		}
		merlinStatusUpdater, err := status.New(statusConfig)
		if err != nil {
			return nil, err
		}
		updaters = append(updaters, merlinStatusUpdater)
	}

	return updaters, nil
}


<|start_filename|>merlin/netlink.go<|end_filename|>
package merlin

import (
	"fmt"
	"strings"

	"github.com/vishvananda/netlink"
)

// netlinkWrapper is an interface to mock out netlink interactions for tests.
type netlinkWrapper interface {
	addVIP(vipInterface, vip string) error
	removeVIP(vipInterface, vip string) error
}

type netlinkWrapperImpl struct{}

func (i *netlinkWrapperImpl) handleVIP(vipInterface, vip string, fn func(netlink.Link, *netlink.Addr) error) error {
	if !strings.Contains(vip, "/") {
		// Doesn't contain a network, add /32.
		vip = vip + "/32"
	}
	ipNet, err := netlink.ParseIPNet(vip)
	if err != nil {
		return fmt.Errorf("unable to parse VIP %s: %v", vip, err)
	}
	lnk, err := netlink.LinkByName(vipInterface)
	if err != nil {
		return fmt.Errorf("unable to add/remove VIP on %s: %v", vipInterface, err)
	}
	if err := fn(lnk, &netlink.Addr{IPNet: ipNet}); err != nil {
		return fmt.Errorf("unable to add/remove VIP %s to %s: %v", vip, vipInterface, err)
	}
	return nil
}

func (i *netlinkWrapperImpl) addVIP(vipInterface, vip string) error {
	return i.handleVIP(vipInterface, vip, netlink.AddrAdd)
}

func (i *netlinkWrapperImpl) removeVIP(vipInterface, vip string) error {
	return i.handleVIP(vipInterface, vip, netlink.AddrDel)
}


<|start_filename|>docker/dns/Dockerfile<|end_filename|>
FROM phusion/baseimage:0.9.19

RUN apt-get update \
    && apt-get upgrade -y -o Dpkg::Options::="--force-confnew" \
	&& apt-get install --no-install-recommends --no-install-suggests -y \
						ca-certificates \
						curl \
						dnsutils \
						vim-tiny \
						lsof \
	&& rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* # 2016-07-13

COPY feed-dns /

ENTRYPOINT ["/sbin/my_init", "--quiet", "--", "/feed-dns"]


<|start_filename|>dns/dns_updater_test.go<|end_filename|>
package dns

import (
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	awselb "github.com/aws/aws-sdk-go/service/elb"
	awsalb "github.com/aws/aws-sdk-go/service/elbv2"
	"github.com/aws/aws-sdk-go/service/route53"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/dns/adapter"
	"github.com/sky-uk/feed/elb"
	"github.com/sky-uk/feed/util/metrics"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
)

func init() {
	metrics.SetConstLabels(make(prometheus.Labels))
	initMetrics()
}

const (
	hostedZoneID                 = "1234"
	domain                       = "james.com."
	lbHostedZoneID               = "lb-hosted-zone-id"
	elbLabelValue                = "elbLabelValue"
	internalALBName              = "internal-alb"
	externalALBName              = "external-alb"
	internalALBDnsName           = "internal-alb-dns-name"
	externalALBDnsName           = "external-alb-dns-name"
	unassocALBDnsName            = "unassoc-alb-dns-name"
	internalALBDnsNameWithPeriod = internalALBDnsName + "."
	externalALBDnsNameWithPeriod = externalALBDnsName + "."
	unassocALBDnsNameWithPeriod  = unassocALBDnsName + "."
	internalScheme               = "internal"
	externalScheme               = "external"
	internalAddressArgument      = "ha-ingress-internal"
	externalAddressArgument      = "ha-ingress-external"
)

var albNames = []string{internalALBName, externalALBName}
var lbDetails = []lbDetail{
	{scheme: internalScheme, dnsName: internalALBDnsName},
	{scheme: externalScheme, dnsName: externalALBDnsName},
}

type lbDetail struct {
	scheme  string
	dnsName string
}

type mockALB struct {
	mock.Mock
}

func (m *mockALB) DescribeLoadBalancers(input *awsalb.DescribeLoadBalancersInput) (*awsalb.DescribeLoadBalancersOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*awsalb.DescribeLoadBalancersOutput), args.Error(1)
}

func (m *mockALB) mockDescribeLoadBalancers(names []string, lbDetails []lbDetail, err error) {
	var lbs []*awsalb.LoadBalancer

	for _, lb := range lbDetails {
		lbs = append(lbs, &awsalb.LoadBalancer{
			Scheme:                aws.String(lb.scheme),
			DNSName:               aws.String(lb.dnsName),
			CanonicalHostedZoneId: aws.String(lbHostedZoneID),
		})
	}

	out := &awsalb.DescribeLoadBalancersOutput{
		LoadBalancers: lbs,
	}

	m.On("DescribeLoadBalancers", &awsalb.DescribeLoadBalancersInput{
		Names: aws.StringSlice(names),
	}).Return(out, err)
}

type mockELB struct {
	mock.Mock
}

func (m *mockELB) FindFrontEndElbs(e elb.ELB, labelValue string) (map[string]elb.LoadBalancerDetails, error) {
	args := m.Called(e, labelValue)
	return args.Get(0).(map[string]elb.LoadBalancerDetails), args.Error(1)
}

func (m *mockELB) mockFindFrontEndElbs(labelValue string, lbDetails []lbDetail, err error) {
	lbs := make(map[string]elb.LoadBalancerDetails)
	if lbDetails != nil {
		for _, lb := range lbDetails {
			lbs[lb.scheme] = elb.LoadBalancerDetails{
				DNSName:      lb.dnsName,
				HostedZoneID: lbHostedZoneID,
			}
		}
	}

	m.On("FindFrontEndElbs", mock.Anything, labelValue).Return(lbs, err)
}

func (m *mockELB) DescribeLoadBalancers(input *awselb.DescribeLoadBalancersInput) (*awselb.DescribeLoadBalancersOutput, error) {
	return nil, nil
}

func (m *mockELB) DescribeTags(input *awselb.DescribeTagsInput) (*awselb.DescribeTagsOutput, error) {
	return nil, nil
}

func (m *mockELB) RegisterInstancesWithLoadBalancer(input *awselb.RegisterInstancesWithLoadBalancerInput) (*awselb.RegisterInstancesWithLoadBalancerOutput, error) {
	return nil, nil
}

func (m *mockELB) DeregisterInstancesFromLoadBalancer(input *awselb.DeregisterInstancesFromLoadBalancerInput) (*awselb.DeregisterInstancesFromLoadBalancerOutput, error) {
	return nil, nil
}

type mockR53Client struct {
	mock.Mock
}

func (m *mockR53Client) GetHostedZoneDomain() (string, error) {
	args := m.Called()
	return args.String(0), args.Error(1)
}

func (m *mockR53Client) UpdateRecordSets(changes []*route53.Change) error {
	args := m.Called(changes)
	return args.Error(0)
}

func (m *mockR53Client) GetRecords() ([]*route53.ResourceRecordSet, error) {
	args := m.Called()
	if args.Error(1) != nil {
		return nil, args.Error(1)
	}

	return args.Get(0).([]*route53.ResourceRecordSet), args.Error(1)
}

func (m *mockR53Client) mockGetRecords(rs []*route53.ResourceRecordSet, err error) {
	m.On("GetRecords").Return(rs, err)
}

func (m *mockR53Client) mockGetHostedZoneDomain() {
	m.On("GetHostedZoneDomain").Return(domain, nil)
}

func setupForELB(albNames []string, elbLabelValue string) (*updater, *mockR53Client, *mockELB, *mockALB) {
	mockALB := &mockALB{}
	mockELB := &mockELB{}

	config := adapter.AWSAdapterConfig{
		HostedZoneID:  hostedZoneID,
		ELBLabelValue: elbLabelValue,
		ALBNames:      albNames,
		ELBClient:     mockELB,
		ALBClient:     mockALB,
		ELBFinder:     mockELB.FindFrontEndElbs,
	}
	lbAdapter, _ := adapter.NewAWSAdapter(&config)
	dnsUpdater := New(hostedZoneID, lbAdapter, 1).(*updater)

	mockR53 := &mockR53Client{}
	dnsUpdater.r53 = mockR53
	return dnsUpdater, mockR53, mockELB, mockALB
}

func setupForExplicitAddresses(definedFrontends map[string]string) (*updater, *mockR53Client) {
	lbAdapter := adapter.NewStaticHostnameAdapter(definedFrontends, 5*time.Minute)

	dnsUpdater := New(hostedZoneID, lbAdapter, 1).(*updater)
	mockR53 := &mockR53Client{}
	dnsUpdater.r53 = mockR53
	return dnsUpdater, mockR53
}

func TestFailsToQueryFrontends(t *testing.T) {
	dnsUpdater, mockR53, _, mockALB := setupForELB(albNames, "")
	mockALB.mockDescribeLoadBalancers(albNames, nil, errors.New("doh"))
	mockR53.mockGetHostedZoneDomain()

	err := dnsUpdater.Start()

	assert.Error(t, err)
}

func TestQueryFrontendElbsFails(t *testing.T) {
	dnsUpdater, mockR53, mockELB, _ := setupForELB(nil, elbLabelValue)
	mockELB.mockFindFrontEndElbs(elbLabelValue, nil, errors.New("nope"))
	mockR53.mockGetHostedZoneDomain()

	err := dnsUpdater.Start()

	assert.EqualError(t, err, "unable to find front end load balancers: nope")
}

func TestQueryFrontendElbsTrailingDotOnDomain(t *testing.T) {
	dnsUpdater, mockR53, mockELB, _ := setupForELB(nil, elbLabelValue)
	lbDetWithDot := []lbDetail{
		{scheme: internalScheme, dnsName: internalALBDnsName},
		{scheme: externalScheme, dnsName: externalALBDnsNameWithPeriod},
	}
	mockELB.mockFindFrontEndElbs(elbLabelValue, lbDetWithDot, nil)
	mockR53.mockGetHostedZoneDomain()

	err := dnsUpdater.Start()

	assert.EqualError(t, err, "unexpected trailing dot on load balancer DNS name: "+externalALBDnsNameWithPeriod)
}

func TestQueryFrontedElbs(t *testing.T) {
	dnsUpdater, mockR53, mockELB, _ := setupForELB(nil, elbLabelValue)
	mockELB.mockFindFrontEndElbs(elbLabelValue, lbDetails, nil)
	mockR53.mockGetHostedZoneDomain()

	assert.NoError(t, dnsUpdater.Start())
	assert.Equal(t, map[string]adapter.DNSDetails{
		internalScheme: {DNSName: internalALBDnsNameWithPeriod, HostedZoneID: lbHostedZoneID},
		externalScheme: {DNSName: externalALBDnsNameWithPeriod, HostedZoneID: lbHostedZoneID},
	}, dnsUpdater.schemeToFrontendMap)
	mockR53.AssertExpectations(t)
}

func TestGetsDomainNameFails(t *testing.T) {
	dnsUpdater, mockR53, _, mockALB := setupForELB(albNames, "")
	mockALB.mockDescribeLoadBalancers(albNames, lbDetails, nil)
	mockR53.On("GetHostedZoneDomain").Return("", errors.New("no domain for you"))

	err := dnsUpdater.Start()

	assert.Error(t, err)
}

func TestUpdateRecordSetFail(t *testing.T) {
	// given
	dnsUpdater, mockR53, _, mockALB := setupForELB(albNames, "")
	mockR53.mockGetHostedZoneDomain()
	mockR53.mockGetRecords(nil, nil)
	mockALB.mockDescribeLoadBalancers(albNames, lbDetails, nil)

	ingressUpdate := []controller.IngressEntry{{Host: "verification.james.com", LbScheme: internalScheme}}

	mockR53.On("UpdateRecordSets", mock.Anything).Return(errors.New("no updates for you"))

	// when
	assert.NoError(t, dnsUpdater.Start())
	err := dnsUpdater.Update(ingressUpdate)

	//then
	assert.Error(t, err)
}

func TestRecordSetUpdates(t *testing.T) {
	var tests = []struct {
		name            string
		update          controller.IngressEntries
		records         []*route53.ResourceRecordSet
		expectedChanges []*route53.Change
	}{
		{
			"Empty update has no change",
			nil, nil, nil,
		},
		{
			"Add new record",
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "cats.james.com",
				Path:        "/",
				LbScheme:    internalScheme,
				ServicePort: 80,
			}},
			nil,
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("cats.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(internalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			}},
		},
		{
			"Updating existing record to a new elb schema",
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{{
				Name: aws.String("foo.james.com."),
				Type: aws.String(route53.RRTypeA),
				AliasTarget: &route53.AliasTarget{
					DNSName:              aws.String(internalALBDnsNameWithPeriod),
					HostedZoneId:         aws.String(lbHostedZoneID),
					EvaluateTargetHealth: aws.Bool(false),
				},
			}},
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(externalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			}},
		},
		{
			"Deleting existing record",
			nil,
			[]*route53.ResourceRecordSet{
				{
					Name: aws.String("foo.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(internalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
				{
					Name: aws.String("bar.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(unassocALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			},
			[]*route53.Change{{
				Action: aws.String("DELETE"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(internalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			}},
		},
		{
			"Adding and deleting records",
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    internalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{
				{
					Name: aws.String("bar.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(internalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
				{
					Name: aws.String("baz.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(unassocALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			},
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(internalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				}},
				{
					Action: aws.String("DELETE"),
					ResourceRecordSet: &route53.ResourceRecordSet{
						Name: aws.String("bar.james.com."),
						Type: aws.String(route53.RRTypeA),
						AliasTarget: &route53.AliasTarget{
							DNSName:              aws.String(internalALBDnsNameWithPeriod),
							HostedZoneId:         aws.String(lbHostedZoneID),
							EvaluateTargetHealth: aws.Bool(false),
						},
					},
				},
			},
		},
		{
			"Non-matching schemes and domains are ignored",
			[]controller.IngressEntry{
				// host doesn't match james.com.
				{
					Name:        "test-entry",
					Host:        "foo.com",
					Path:        "/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
				// scheme doesn't match internal
				{
					Name:        "test-entry",
					Host:        "foo.james.com",
					Path:        "/",
					LbScheme:    "invalidscheme",
					ServicePort: 80,
				},
			},
			nil,
			nil,
		},
		{
			"Duplicate hosts are not duplicated in changeset",
			[]controller.IngressEntry{
				{
					Name:        "test-entry",
					Host:        "foo.james.com",
					Path:        "/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
				{
					Name:        "test-entry-blah",
					Host:        "foo.james.com",
					Path:        "/blah/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
				{
					Name:        "test-entry-lala",
					Host:        "foo.james.com",
					Path:        "/lala/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
			},
			nil,
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(internalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			}},
		},
		{
			"Should choose first conflicting scheme",
			[]controller.IngressEntry{
				{
					Name:        "test-entry",
					Host:        "bar.james.com",
					Path:        "/",
					LbScheme:    externalScheme,
					ServicePort: 80,
				},
				{
					Name:        "test-entry",
					Host:        "bar.james.com",
					Path:        "/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
			},
			nil,
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("bar.james.com."),
					Type: aws.String(route53.RRTypeA),
					AliasTarget: &route53.AliasTarget{
						DNSName:              aws.String(externalALBDnsNameWithPeriod),
						HostedZoneId:         aws.String(lbHostedZoneID),
						EvaluateTargetHealth: aws.Bool(false),
					},
				},
			}},
		},
		{
			"Does not update records when current and new entry are the same",
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{{
				Name: aws.String("foo.james.com."),
				Type: aws.String(route53.RRTypeA),
				AliasTarget: &route53.AliasTarget{
					DNSName:              aws.String(externalALBDnsNameWithPeriod),
					HostedZoneId:         aws.String(lbHostedZoneID),
					EvaluateTargetHealth: aws.Bool(false),
				},
			}},
			nil,
		},
	}

	for _, test := range tests {
		fmt.Printf("=== test: %s\n", test.name)

		dnsUpdater, mockR53, _, mockALB := setupForELB(albNames, "")
		mockALB.mockDescribeLoadBalancers(albNames, lbDetails, nil)
		mockR53.mockGetHostedZoneDomain()
		mockR53.mockGetRecords(test.records, nil)
		mockR53.On("UpdateRecordSets", test.expectedChanges).Return(nil)

		assert.NoError(t, dnsUpdater.Start())
		assert.NoError(t, dnsUpdater.Update(test.update))

		mockR53.AssertExpectations(t)

		if t.Failed() {
			t.FailNow()
		}
	}
}

func TestRecordSetUpdatesWithAddressArguments(t *testing.T) {
	ttl := aws.Int64(300)
	internalAndExternalFrontends := map[string]string{internalScheme: internalAddressArgument, externalScheme: externalAddressArgument}

	var tests = []struct {
		name             string
		definedFrontends map[string]string
		update           controller.IngressEntries
		records          []*route53.ResourceRecordSet
		expectedChanges  []*route53.Change
	}{
		{
			"Empty update has no change",
			internalAndExternalFrontends,
			controller.IngressEntries{},
			nil,
			nil,
		},
		{
			"Add new record",
			internalAndExternalFrontends,
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "cats.james.com",
				Path:        "/",
				LbScheme:    internalScheme,
				ServicePort: 80,
			}},
			nil,
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("cats.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(internalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Updating existing record to a new scheme",
			internalAndExternalFrontends,
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{{
				Name: aws.String("foo.james.com."),
				Type: aws.String(route53.RRTypeCname),
				ResourceRecords: []*route53.ResourceRecord{
					{
						Value: aws.String(internalAddressArgument),
					},
				},
				TTL: ttl,
			}},
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(externalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Deleting existing record",
			internalAndExternalFrontends,
			controller.IngressEntries{},
			[]*route53.ResourceRecordSet{
				{
					Name: aws.String("foo.com."),
					Type: aws.String(route53.RRTypeCname),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(internalAddressArgument),
						},
					},
					TTL: ttl,
				},
				{
					Name: aws.String("bar.com."),
					Type: aws.String(route53.RRTypeCname),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String("some-ingress-we-dont-manage.james.com"),
						},
					},
					TTL: ttl,
				},
			},
			[]*route53.Change{{
				Action: aws.String("DELETE"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(internalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Adding and deleting records",
			internalAndExternalFrontends,
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    internalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{
				{
					Name: aws.String("bar.james.com."),
					Type: aws.String(route53.RRTypeCname),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(internalAddressArgument),
						},
					},
					TTL: ttl,
				},
				{
					Name: aws.String("baz.james.com."),
					Type: aws.String(route53.RRTypeCname),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String("somerandom-ingress.s.sandbox.james.com"),
						},
					},
					TTL: ttl,
				},
			},
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(internalAddressArgument),
						},
					},
					TTL: ttl,
				}},
				{
					Action: aws.String("DELETE"),
					ResourceRecordSet: &route53.ResourceRecordSet{
						Name: aws.String("bar.james.com."),
						Type: aws.String("CNAME"),
						ResourceRecords: []*route53.ResourceRecord{
							{
								Value: aws.String(internalAddressArgument),
							},
						},
						TTL: ttl,
					},
				},
			},
		},
		{
			"Non-matching schemes and domains are ignored",
			internalAndExternalFrontends,
			[]controller.IngressEntry{
				// host doesn't match james.com.
				{
					Name:        "test-entry",
					Host:        "foo.com",
					Path:        "/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
				// scheme doesn't match internal
				{
					Name:        "test-entry",
					Host:        "foo.james.com",
					Path:        "/",
					LbScheme:    "invalidscheme",
					ServicePort: 80,
				},
			},
			nil,
			nil,
		},
		{
			"Duplicate hosts are not duplicated in changeset",
			internalAndExternalFrontends,
			[]controller.IngressEntry{
				{
					Name:        "test-entry",
					Host:        "foo.james.com",
					Path:        "/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
				{
					Name:        "test-entry-blah",
					Host:        "foo.james.com",
					Path:        "/blah/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
				{
					Name:        "test-entry-lala",
					Host:        "foo.james.com",
					Path:        "/lala/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
			},
			nil,
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(internalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Should choose first conflicting scheme",
			internalAndExternalFrontends,
			[]controller.IngressEntry{
				{
					Name:        "test-entry",
					Host:        "bar.james.com",
					Path:        "/",
					LbScheme:    externalScheme,
					ServicePort: 80,
				},
				{
					Name:        "test-entry",
					Host:        "bar.james.com",
					Path:        "/",
					LbScheme:    internalScheme,
					ServicePort: 80,
				},
			},
			nil,
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("bar.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(externalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Does not update records when current and new entry are the same",
			internalAndExternalFrontends,
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{{
				Name: aws.String("foo.james.com."),
				Type: aws.String(route53.RRTypeCname),
				TTL:  ttl,
				ResourceRecords: []*route53.ResourceRecord{
					{
						Value: aws.String(externalAddressArgument),
					},
				},
			}},
			nil,
		},
		{
			"Updates TTL on a CNAME record when it has changed",
			internalAndExternalFrontends,
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{{
				Name: aws.String("foo.james.com."),
				Type: aws.String(route53.RRTypeCname),
				TTL:  aws.Int64(999),
				ResourceRecords: []*route53.ResourceRecord{
					{
						Value: aws.String(externalAddressArgument),
					},
				},
			}},
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(externalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Handles existing records which have no TTL",
			internalAndExternalFrontends,
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			[]*route53.ResourceRecordSet{{
				Name: aws.String("foo.james.com."),
				Type: aws.String(route53.RRTypeCname),
				ResourceRecords: []*route53.ResourceRecord{
					{
						Value: aws.String(externalAddressArgument),
					},
				},
			}},
			[]*route53.Change{{
				Action: aws.String("UPSERT"),
				ResourceRecordSet: &route53.ResourceRecordSet{
					Name: aws.String("foo.james.com."),
					Type: aws.String("CNAME"),
					ResourceRecords: []*route53.ResourceRecord{
						{
							Value: aws.String(externalAddressArgument),
						},
					},
					TTL: ttl,
				},
			}},
		},
		{
			"Ignores ingresses which use a scheme for which no frontend is defined",
			map[string]string{internalScheme: internalAddressArgument},
			[]controller.IngressEntry{{
				Name:        "test-entry",
				Host:        "foo.james.com",
				Path:        "/",
				LbScheme:    externalScheme,
				ServicePort: 80,
			}},
			nil,
			nil,
		},
	}

	for _, test := range tests {
		fmt.Printf("=== test: TestRecordSetUpdatesWithAddressArguments: %s\n", test.name)

		dnsUpdater, mockR53 := setupForExplicitAddresses(test.definedFrontends)
		mockR53.mockGetHostedZoneDomain()
		mockR53.mockGetRecords(test.records, nil)
		mockR53.On("UpdateRecordSets", test.expectedChanges).Return(nil)

		assert.NoError(t, dnsUpdater.Start())
		assert.NoError(t, dnsUpdater.Update(test.update))

		mockR53.AssertExpectations(t)

		if t.Failed() {
			t.FailNow()
		}
	}
}


<|start_filename|>elb/elb_test.go<|end_filename|>
package elb

import (
	"errors"
	"fmt"
	"reflect"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/ec2metadata"
	awselb "github.com/aws/aws-sdk-go/service/elb"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/util/metrics"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
)

func init() {
	metrics.SetConstLabels(make(prometheus.Labels))
}

const (
	clusterName               = "cluster_name"
	ingressName               = "ingress_name"
	region                    = "eu-west-1"
	frontendTag               = "sky.uk/KubernetesClusterFrontend"
	ingressNameTag            = "sky.uk/KubernetesClusterIngressClass"
	canonicalHostedZoneNameID = "test-id"
	elbDNSName                = "elb-dnsname"
	elbInternalScheme         = "internal"
	elbInternetFacingScheme   = "internet-facing"
)

var defaultTags = []*awselb.Tag{
	{Key: aws.String(frontendTag), Value: aws.String(clusterName)},
	{Key: aws.String(ingressNameTag), Value: aws.String(ingressName)},
}

type fakeElb struct {
	mock.Mock
}

func (m *fakeElb) DescribeLoadBalancers(input *awselb.DescribeLoadBalancersInput) (*awselb.DescribeLoadBalancersOutput, error) {
	args := m.Called(input)

	return args.Get(0).(*awselb.DescribeLoadBalancersOutput), args.Error(1)
}

func (m *fakeElb) DescribeTags(input *awselb.DescribeTagsInput) (*awselb.DescribeTagsOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*awselb.DescribeTagsOutput), args.Error(1)
}

func (m *fakeElb) DeregisterInstancesFromLoadBalancer(input *awselb.DeregisterInstancesFromLoadBalancerInput) (*awselb.DeregisterInstancesFromLoadBalancerOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*awselb.DeregisterInstancesFromLoadBalancerOutput), args.Error(1)
}

func (m *fakeElb) RegisterInstancesWithLoadBalancer(input *awselb.RegisterInstancesWithLoadBalancerInput) (*awselb.RegisterInstancesWithLoadBalancerOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*awselb.RegisterInstancesWithLoadBalancerOutput), args.Error(1)
}

type fakeMetadata struct {
	mock.Mock
}

func (m *fakeMetadata) Available() bool {
	args := m.Called()
	return args.Bool(0)
}

func (m *fakeMetadata) Region() (string, error) {
	args := m.Called()
	return args.String(0), nil
}

func (m *fakeMetadata) GetInstanceIdentityDocument() (ec2metadata.EC2InstanceIdentityDocument, error) {
	args := m.Called()
	return args.Get(0).(ec2metadata.EC2InstanceIdentityDocument), args.Error(1)
}

type lb struct {
	name   string
	scheme string
}

func mockLoadBalancers(m *fakeElb, lbs ...lb) {
	var descriptions []*awselb.LoadBalancerDescription
	for _, lb := range lbs {
		descriptions = append(descriptions, &awselb.LoadBalancerDescription{
			LoadBalancerName:          aws.String(lb.name),
			CanonicalHostedZoneNameID: aws.String(canonicalHostedZoneNameID),
			Scheme:                    aws.String(lb.scheme),
			DNSName:                   aws.String(elbDNSName),
		})

	}
	m.On("DescribeLoadBalancers", mock.AnythingOfType("*elb.DescribeLoadBalancersInput")).Return(&awselb.DescribeLoadBalancersOutput{
		LoadBalancerDescriptions: descriptions,
	}, nil)
}

type lbTags struct {
	tags []*awselb.Tag
	name string
}

func mockClusterTags(m *fakeElb, lbs ...lbTags) {
	var tagDescriptions []*awselb.TagDescription

	for _, lb := range lbs {
		tagDescriptions = append(tagDescriptions, &awselb.TagDescription{
			LoadBalancerName: aws.String(lb.name),
			Tags:             lb.tags,
		})
	}

	m.On("DescribeTags", mock.AnythingOfType("*elb.DescribeTagsInput")).Return(&awselb.DescribeTagsOutput{
		TagDescriptions: tagDescriptions,
	}, nil)
}

func mockRegisterInstances(mockElb *fakeElb, elbName, instanceID string) {
	mockElb.On("RegisterInstancesWithLoadBalancer", &awselb.RegisterInstancesWithLoadBalancerInput{
		LoadBalancerName: aws.String(elbName),
		Instances:        []*awselb.Instance{{InstanceId: aws.String(instanceID)}},
	}).Return(&awselb.RegisterInstancesWithLoadBalancerOutput{
		Instances: []*awselb.Instance{{InstanceId: aws.String(instanceID)}},
	}, nil)
}

func mockInstanceMetadata(mockMd *fakeMetadata, instanceID string) {
	mockMd.On("GetInstanceIdentityDocument").Return(ec2metadata.EC2InstanceIdentityDocument{InstanceID: instanceID}, nil)
}

func setup() (controller.Updater, *fakeElb, *fakeMetadata) {
	e, _ := New(region, clusterName, ingressName, 1, 0)
	mockElb := &fakeElb{}
	mockMetadata := &fakeMetadata{}
	e.(*elb).awsElb = mockElb
	e.(*elb).metadata = mockMetadata
	return e, mockElb, mockMetadata
}

func TestMetricsRegisteredCorrectly(t *testing.T) {
	//when
	_, _ = New(region, clusterName, ingressName, 1, 0)

	//then
	assert.Equal(t, "feed_ingress_frontends_attached", metricName(attachedFrontendGauge))
}

func metricName(c prometheus.Collector) string {
	descriptionCh := make(chan *prometheus.Desc, 1)
	c.Describe(descriptionCh)
	desc := <-descriptionCh
	descReflect := reflect.ValueOf(*desc)
	fqNameField := descReflect.FieldByName("fqName")
	return fqNameField.String()
}

func TestCanNotCreateUpdaterWithoutFrontEndTagValue(t *testing.T) {
	//when
	_, err := New(region, "", ingressName, 1, 0)

	//then
	assert.Error(t, err)
}

func TestCanNotCreateUpdaterWithoutIngressNameTagValue(t *testing.T) {
	//when
	_, err := New(region, clusterName, "", 1, 0)

	//then
	assert.Error(t, err)
}

func TestAttachWithSingleMatchingLoadBalancers(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	clusterFrontEndDifferentCluster := "cluster-frontend-different-cluster"
	mockLoadBalancers(mockElb,
		lb{clusterFrontEnd, elbInternalScheme},
		lb{clusterFrontEndDifferentCluster, elbInternalScheme},
		lb{"other", elbInternalScheme})

	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags},
		lbTags{name: clusterFrontEndDifferentCluster, tags: []*awselb.Tag{
			{Key: aws.String(frontendTag), Value: aws.String("different cluster")},
			{Key: aws.String(ingressName), Value: aws.String("different cluster")},
		}},
		lbTags{name: "other elb", tags: []*awselb.Tag{{Key: aws.String("Banana"), Value: aws.String("Tasty")}}},
	)
	mockRegisterInstances(mockElb, clusterFrontEnd, instanceID)
	err := e.Start()

	//when
	_ = e.Update(controller.IngressEntries{})

	//then
	assert.NoError(t, e.Health())
	mockElb.AssertExpectations(t)
	mockMetadata.AssertExpectations(t)
	assert.NoError(t, err)
}

func TestReportsErrorIfExpectedNotMatched(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	e.(*elb).expectedNumber = 2
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	clusterFrontEndDifferentCluster := "cluster-frontend-different-cluster"
	mockLoadBalancers(mockElb,
		lb{name: clusterFrontEnd, scheme: elbInternalScheme},
		lb{name: clusterFrontEndDifferentCluster, scheme: elbInternalScheme},
		lb{name: "other", scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags},
		lbTags{name: clusterFrontEndDifferentCluster, tags: []*awselb.Tag{
			{Key: aws.String(frontendTag), Value: aws.String("different cluster")},
			{Key: aws.String(ingressNameTag), Value: aws.String("different cluster")},
		}},
		lbTags{name: "other elb", tags: []*awselb.Tag{{Key: aws.String("Banana"), Value: aws.String("Tasty")}}},
	)
	mockRegisterInstances(mockElb, clusterFrontEnd, instanceID)

	//when
	_ = e.Start()
	err := e.Update(controller.IngressEntries{})

	//then
	assert.EqualError(t, err, "expected ELBs: 2 actual: 1")
}

func TestNameAndDNSNameAndHostedZoneIDLoadBalancerDetailsAreExtracted(t *testing.T) {
	//given
	mockElb := &fakeElb{}
	clusterFrontEnd := "cluster-frontend"
	mockLoadBalancers(mockElb, lb{name: clusterFrontEnd, scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags},
	)

	//when
	frontends, _ := FindFrontEndElbsWithIngressClassName(mockElb, clusterName, ingressName)

	//then
	assert.Equal(t, "cluster-frontend", frontends[elbInternalScheme].Name)
	assert.Equal(t, elbDNSName, frontends[elbInternalScheme].DNSName)
	assert.Equal(t, canonicalHostedZoneNameID, frontends[elbInternalScheme].HostedZoneID)
	assert.Equal(t, elbInternalScheme, frontends[elbInternalScheme].Scheme)
}

func TestFindElbWithoutIngressName(t *testing.T) {
	//given
	mockElb := &fakeElb{}
	clusterFrontEnd := "cluster-frontend"
	mockLoadBalancers(mockElb, lb{name: clusterFrontEnd, scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: []*awselb.Tag{
			{Key: aws.String(frontendTag), Value: aws.String(clusterName)},
		}},
	)

	//when
	frontends, _ := FindFrontEndElbs(mockElb, clusterName)

	//then
	assert.Equal(t, "cluster-frontend", frontends[elbInternalScheme].Name)
	assert.Equal(t, elbDNSName, frontends[elbInternalScheme].DNSName)
	assert.Equal(t, canonicalHostedZoneNameID, frontends[elbInternalScheme].HostedZoneID)
	assert.Equal(t, elbInternalScheme, frontends[elbInternalScheme].Scheme)
}

func TestAttachWithInternalAndInternetFacing(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	e.(*elb).expectedNumber = 2
	instanceID := "cow"
	privateFrontend := "cluster-frontend"
	publicFrontend := "cluster-frontend2"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockLoadBalancers(mockElb,
		lb{name: privateFrontend, scheme: elbInternalScheme},
		lb{name: publicFrontend, scheme: elbInternetFacingScheme})
	mockClusterTags(mockElb,
		lbTags{name: privateFrontend, tags: defaultTags},
		lbTags{name: publicFrontend, tags: defaultTags},
	)
	mockRegisterInstances(mockElb, privateFrontend, instanceID)
	mockRegisterInstances(mockElb, publicFrontend, instanceID)

	//when
	err := e.Start()
	_ = e.Update(controller.IngressEntries{})

	//then
	mockElb.AssertExpectations(t)
	mockMetadata.AssertExpectations(t)
	assert.NoError(t, err)
}

func TestErrorGettingMetadata(t *testing.T) {
	e, _, mockMetadata := setup()
	mockMetadata.On("GetInstanceIdentityDocument").Return(ec2metadata.EC2InstanceIdentityDocument{}, fmt.Errorf("no metadata for you"))

	err := e.Update(controller.IngressEntries{})

	assert.EqualError(t, err, "unable to query ec2 metadata service for InstanceId: no metadata for you")
}

func TestErrorDescribingInstances(t *testing.T) {
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockElb.On("DescribeLoadBalancers", mock.AnythingOfType("*elb.DescribeLoadBalancersInput")).Return(&awselb.DescribeLoadBalancersOutput{}, errors.New("oh dear oh dear"))

	_ = e.Start()
	err := e.Update(controller.IngressEntries{})

	assert.EqualError(t, err, "unable to describe load balancers: oh dear oh dear")
}

func TestErrorDescribingTags(t *testing.T) {
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockLoadBalancers(mockElb, lb{name: "one"})
	mockElb.On("DescribeTags", mock.AnythingOfType("*elb.DescribeTagsInput")).Return(&awselb.DescribeTagsOutput{}, errors.New("oh dear oh dear"))

	_ = e.Start()
	err := e.Update(controller.IngressEntries{})

	assert.EqualError(t, err, "unable to describe tags: oh dear oh dear")
}

func TestNoMatchingElbs(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	loadBalancerName := "i am not the loadbalancer you are looking for"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockLoadBalancers(mockElb, lb{name: loadBalancerName, scheme: elbInternalScheme})
	// No cluster tags
	mockClusterTags(mockElb, lbTags{name: loadBalancerName, tags: []*awselb.Tag{}})

	// when
	_ = e.Start()
	err := e.Update(controller.IngressEntries{})

	// then
	assert.Error(t, err, "expected ELBs: 1 actual: 0")
}

func TestAttachingWithoutIngressNameTagElbs(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	loadBalancerName := "i am not the loadbalancer you are looking for"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockLoadBalancers(mockElb, lb{name: loadBalancerName, scheme: elbInternalScheme})
	// No cluster tags
	mockClusterTags(mockElb, lbTags{name: loadBalancerName, tags: []*awselb.Tag{
		{Key: aws.String(frontendTag), Value: aws.String(clusterName)},
	}})

	// when
	_ = e.Start()
	err := e.Update(controller.IngressEntries{})

	// then
	assert.Error(t, err, "expected ELBs: 1 actual: 0")
}

func TestAttachingWithoutFrontendTagElbs(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	loadBalancerName := "i am not the loadbalancer you are looking for"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockLoadBalancers(mockElb, lb{name: loadBalancerName, scheme: elbInternalScheme})
	// No cluster tags
	mockClusterTags(mockElb, lbTags{name: loadBalancerName, tags: []*awselb.Tag{
		{Key: aws.String(ingressNameTag), Value: aws.String(ingressName)},
	}})

	// when
	_ = e.Start()
	err := e.Update(controller.IngressEntries{})

	// then
	assert.Error(t, err, "expected ELBs: 1 actual: 0")
}

func TestGetLoadBalancerPages(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	loadBalancerName := "lb1"
	mockElb.On("DescribeLoadBalancers", &awselb.DescribeLoadBalancersInput{}).Return(&awselb.DescribeLoadBalancersOutput{NextMarker: aws.String("Use me")}, nil)
	mockElb.On("DescribeLoadBalancers", &awselb.DescribeLoadBalancersInput{Marker: aws.String("Use me")}).Return(&awselb.DescribeLoadBalancersOutput{
		LoadBalancerDescriptions: []*awselb.LoadBalancerDescription{{
			LoadBalancerName:          aws.String(loadBalancerName),
			DNSName:                   aws.String(elbDNSName),
			CanonicalHostedZoneNameID: aws.String(canonicalHostedZoneNameID),
		}},
	}, nil)
	mockInstanceMetadata(mockMetadata, instanceID)
	mockClusterTags(mockElb, lbTags{name: loadBalancerName, tags: defaultTags})
	mockRegisterInstances(mockElb, loadBalancerName, instanceID)

	// when
	err := e.Update(controller.IngressEntries{})

	// then
	assert.NoError(t, err)
	mockElb.AssertExpectations(t)
}

func TestTagCallsPage(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	e.(*elb).expectedNumber = 2
	instanceID := "cow"
	loadBalancerName1 := "lb1"
	loadBalancerName2 := "lb2"
	mockInstanceMetadata(mockMetadata, instanceID)
	mockLoadBalancers(mockElb,
		lb{name: loadBalancerName1, scheme: elbInternalScheme},
		lb{name: loadBalancerName2, scheme: elbInternetFacingScheme})
	mockClusterTags(mockElb,
		lbTags{name: loadBalancerName1, tags: defaultTags},
		lbTags{name: loadBalancerName2, tags: defaultTags})
	mockRegisterInstances(mockElb, loadBalancerName1, instanceID)
	mockRegisterInstances(mockElb, loadBalancerName2, instanceID)

	// when
	err := e.Update(controller.IngressEntries{})

	// then
	assert.NoError(t, err)
	mockElb.AssertExpectations(t)
}

func TestDeregistersWithAttachedELBs(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	e.(*elb).expectedNumber = 2
	e.(*elb).drainDelay = time.Millisecond * 100

	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	clusterFrontEnd2 := "cluster-frontend2"
	mockLoadBalancers(mockElb,
		lb{name: clusterFrontEnd, scheme: elbInternalScheme},
		lb{name: clusterFrontEnd2, scheme: elbInternetFacingScheme},
		lb{name: "other", scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags},
		lbTags{name: clusterFrontEnd2, tags: defaultTags},
		lbTags{name: "other elb", tags: []*awselb.Tag{{Key: aws.String("Banana"), Value: aws.String("Tasty")}}},
	)
	mockRegisterInstances(mockElb, clusterFrontEnd, instanceID)
	mockRegisterInstances(mockElb, clusterFrontEnd2, instanceID)

	mockElb.On("DeregisterInstancesFromLoadBalancer", &awselb.DeregisterInstancesFromLoadBalancerInput{
		Instances:        []*awselb.Instance{{InstanceId: aws.String(instanceID)}},
		LoadBalancerName: aws.String(clusterFrontEnd),
	}).Return(&awselb.DeregisterInstancesFromLoadBalancerOutput{
		Instances: []*awselb.Instance{{InstanceId: aws.String(instanceID)}},
	}, nil)
	mockElb.On("DeregisterInstancesFromLoadBalancer", &awselb.DeregisterInstancesFromLoadBalancerInput{
		Instances:        []*awselb.Instance{{InstanceId: aws.String(instanceID)}},
		LoadBalancerName: aws.String(clusterFrontEnd2),
	}).Return(&awselb.DeregisterInstancesFromLoadBalancerOutput{
		Instances: []*awselb.Instance{{InstanceId: aws.String(instanceID)}},
	}, nil)

	//when
	assert.NoError(t, e.Start())
	assert.NoError(t, e.Update(controller.IngressEntries{}))
	beforeStop := time.Now()
	assert.NoError(t, e.Stop())
	stopDuration := time.Now().Sub(beforeStop)

	//then
	mockElb.AssertExpectations(t)
	assert.True(t, stopDuration.Nanoseconds() > time.Millisecond.Nanoseconds()*50,
		"Drain time should have caused stop to take at least 50ms.")
}

func TestRegisterInstanceError(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	mockLoadBalancers(mockElb, lb{name: clusterFrontEnd, scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags},
	)
	mockElb.On("RegisterInstancesWithLoadBalancer", mock.Anything).Return(&awselb.RegisterInstancesWithLoadBalancerOutput{}, errors.New("no register for you"))

	// when
	err := e.Update(controller.IngressEntries{})

	// then
	assert.EqualError(t, err, "unable to register instance cow with elb cluster-frontend: no register for you")
}

func TestDeRegisterInstanceError(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	mockLoadBalancers(mockElb,
		lb{name: clusterFrontEnd, scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags},
	)
	mockRegisterInstances(mockElb, clusterFrontEnd, instanceID)
	mockElb.On("DeregisterInstancesFromLoadBalancer", mock.Anything).Return(&awselb.DeregisterInstancesFromLoadBalancerOutput{}, errors.New("no deregister for you"))

	// when
	_ = e.Start()
	_ = e.Update(controller.IngressEntries{})
	err := e.Stop()

	// then
	assert.EqualError(t, err, "at least one ELB failed to detach")
}

func TestRetriesUpdateIfFirstAttemptFails(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	mockLoadBalancers(mockElb,
		lb{name: clusterFrontEnd, scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{
			name: clusterFrontEnd,
			tags: defaultTags})
	mockElb.On("RegisterInstancesWithLoadBalancer", mock.Anything).Return(
		&awselb.RegisterInstancesWithLoadBalancerOutput{}, errors.New("no register for you"))

	// when
	_ = e.Start()
	firstErr := e.Update(controller.IngressEntries{})
	secondErr := e.Update(controller.IngressEntries{})

	// then
	assert.Error(t, firstErr)
	assert.Error(t, secondErr)
}

func TestHealthReportsHealthyBeforeFirstUpdate(t *testing.T) {
	// given
	e, _, _ := setup()

	// when
	err := e.Start()

	// then
	assert.NoError(t, err)
	assert.Nil(t, e.Health())
}

func TestHealthReportsUnhealthyAfterUnsuccessfulFirstUpdate(t *testing.T) {
	// given
	e, mockElb, mockMetadata := setup()
	e.(*elb).expectedNumber = 2

	// and
	instanceID := "cow"
	mockInstanceMetadata(mockMetadata, instanceID)
	clusterFrontEnd := "cluster-frontend"
	mockLoadBalancers(mockElb,
		lb{name: clusterFrontEnd, scheme: elbInternalScheme})
	mockClusterTags(mockElb,
		lbTags{name: clusterFrontEnd, tags: defaultTags})
	mockRegisterInstances(mockElb, clusterFrontEnd, instanceID)

	// when
	err := e.Start()
	updateErr := e.Update(controller.IngressEntries{})

	// then
	assert.NoError(t, err)
	assert.Error(t, updateErr)
	assert.Error(t, e.Health())
}


<|start_filename|>dns/adapter/aws_adapter.go<|end_filename|>
package adapter

import (
	"fmt"
	"strings"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	awselb "github.com/aws/aws-sdk-go/service/elb"
	awsalb "github.com/aws/aws-sdk-go/service/elbv2"
	"github.com/aws/aws-sdk-go/service/route53"
	"github.com/sky-uk/feed/elb"
)

// FindELBsFunc defines a function which find ELBs based on a tag value
type FindELBsFunc func(elb elb.ELB, tagValue string) (map[string]elb.LoadBalancerDetails, error)

// ALB represents the subset of AWS operations needed for dns_updater.go
type ALB interface {
	DescribeLoadBalancers(input *awsalb.DescribeLoadBalancersInput) (*awsalb.DescribeLoadBalancersOutput, error)
}

// AWSAdapterConfig describes the configuration of a FrontendAdapter which uses AWS ELBs and/or ALBs
type AWSAdapterConfig struct {
	Region        string
	HostedZoneID  string
	ELBLabelValue string
	ALBNames      []string
	ALBClient     ALB
	ELBClient     elb.ELB
	ELBFinder     FindELBsFunc
}

type awsAdapter struct {
	hostedZoneID     *string
	elbLabelValue    string
	albNames         []string
	elb              elb.ELB
	alb              ALB
	findFrontEndElbs FindELBsFunc
}

// NewAWSAdapter creates a FrontendAdapter which interacts with AWS ELBs or ALBs.
func NewAWSAdapter(config *AWSAdapterConfig) (FrontendAdapter, error) {
	if config.ALBClient == nil && config.ELBClient == nil {
		awsSession, err := session.NewSession(&aws.Config{Region: &config.Region})
		if err != nil {
			return nil, fmt.Errorf("unable to open AWS session: %v", err)
		}

		config.ALBClient = awsalb.New(awsSession)
		config.ELBClient = awselb.New(awsSession)
	}

	if config.ELBFinder == nil {
		config.ELBFinder = elb.FindFrontEndElbs
	}

	return &awsAdapter{
		hostedZoneID:     aws.String(config.HostedZoneID),
		elbLabelValue:    config.ELBLabelValue,
		albNames:         config.ALBNames,
		elb:              config.ELBClient,
		alb:              config.ALBClient,
		findFrontEndElbs: config.ELBFinder,
	}, nil
}

func (a *awsAdapter) Initialise() (map[string]DNSDetails, error) {
	if a.elbLabelValue != "" && len(a.albNames) > 0 {
		return nil, fmt.Errorf("can't specify both elb label value (%s) and alb names (%v) - only one or the other may be"+
			" specified", a.elbLabelValue, a.albNames)
	}

	schemeToFrontendMap := make(map[string]DNSDetails)
	if err := a.initELBs(schemeToFrontendMap); err != nil {
		return nil, err
	}

	if err := a.initALBs(schemeToFrontendMap); err != nil {
		return nil, err
	}

	return schemeToFrontendMap, nil
}

func (a *awsAdapter) initELBs(schemeToFrontendMap map[string]DNSDetails) error {
	if a.elbLabelValue == "" {
		return nil
	}

	elbs, err := a.findFrontEndElbs(a.elb, a.elbLabelValue)
	if err != nil {
		return fmt.Errorf("unable to find front end load balancers: %v", err)
	}

	for scheme, lbDetails := range elbs {
		if strings.HasSuffix(lbDetails.DNSName, ".") {
			return fmt.Errorf("unexpected trailing dot on load balancer DNS name: %s", lbDetails.DNSName)
		}

		schemeToFrontendMap[scheme] = DNSDetails{DNSName: lbDetails.DNSName + ".", HostedZoneID: lbDetails.HostedZoneID}
	}

	return nil
}

func (a *awsAdapter) initALBs(schemeToFrontendMap map[string]DNSDetails) error {
	if len(a.albNames) == 0 {
		return nil
	}

	req := &awsalb.DescribeLoadBalancersInput{Names: aws.StringSlice(a.albNames)}

	for {
		resp, err := a.alb.DescribeLoadBalancers(req)
		if err != nil {
			return err
		}

		for _, lb := range resp.LoadBalancers {
			schemeToFrontendMap[*lb.Scheme] = DNSDetails{DNSName: *lb.DNSName + ".", HostedZoneID: *lb.CanonicalHostedZoneId}
		}

		if resp.NextMarker == nil {
			break
		}

		req.Marker = resp.NextMarker
	}

	return nil
}

func (a *awsAdapter) CreateChange(action string, host string, details DNSDetails, recordExists bool, existingRecord *ConsolidatedRecord) *route53.Change {
	if !recordExists {
		set := &route53.ResourceRecordSet{
			Name: aws.String(host),
		}

		set.Type = aws.String("A")
		set.AliasTarget = &route53.AliasTarget{
			DNSName:      aws.String(details.DNSName),
			HostedZoneId: aws.String(details.HostedZoneID),
			// disable this since we only point to a single load balancer
			EvaluateTargetHealth: aws.Bool(false),
		}

		return &route53.Change{
			Action:            aws.String(action),
			ResourceRecordSet: set,
		}
	}

	return nil
}

func (a *awsAdapter) IsManaged(rrs *route53.ResourceRecordSet) (*ConsolidatedRecord, bool) {
	if *rrs.Type == route53.RRTypeA && rrs.AliasTarget != nil {
		return &ConsolidatedRecord{
			Name:            *rrs.Name,
			PointsTo:        *rrs.AliasTarget.DNSName,
			AliasHostedZone: *rrs.AliasTarget.HostedZoneId,
		}, true
	}

	return nil, false
}


<|start_filename|>alb/alb_test.go<|end_filename|>
package alb

import (
	"errors"
	"reflect"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/ec2metadata"
	aws_alb "github.com/aws/aws-sdk-go/service/elbv2"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/util/metrics"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
)

func init() {
	metrics.SetConstLabels(make(prometheus.Labels))
}

const (
	region = "eu-west-1"
)

type mockALB struct {
	mock.Mock
}

func (m *mockALB) DescribeTargetGroups(input *aws_alb.DescribeTargetGroupsInput) (*aws_alb.DescribeTargetGroupsOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*aws_alb.DescribeTargetGroupsOutput), args.Error(1)
}

func (m *mockALB) RegisterTargets(input *aws_alb.RegisterTargetsInput) (*aws_alb.RegisterTargetsOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*aws_alb.RegisterTargetsOutput), args.Error(1)
}

func (m *mockALB) DeregisterTargets(input *aws_alb.DeregisterTargetsInput) (*aws_alb.DeregisterTargetsOutput, error) {
	args := m.Called(input)
	return args.Get(0).(*aws_alb.DeregisterTargetsOutput), args.Error(1)
}

type mockMetadata struct {
	mock.Mock
}

func (m *mockMetadata) Available() bool {
	args := m.Called()
	return args.Bool(0)
}

func (m *mockMetadata) Region() (string, error) {
	args := m.Called()
	return args.String(0), nil
}

func (m *mockMetadata) GetInstanceIdentityDocument() (ec2metadata.EC2InstanceIdentityDocument, error) {
	args := m.Called()
	return args.Get(0).(ec2metadata.EC2InstanceIdentityDocument), args.Error(1)
}

func (m *mockALB) mockDescribeTargetGroups(names []string, arns []string, reqMarker *string, nextMarker *string, err error) {
	var awsNames []*string
	var targetGroups []*aws_alb.TargetGroup
	for i := range names {
		awsNames = append(awsNames, aws.String(names[i]))
		arn := arns[i]
		if arn != "" {
			targetGroups = append(targetGroups, &aws_alb.TargetGroup{
				TargetGroupArn: aws.String(arn),
			})
		}
	}

	m.On("DescribeTargetGroups",
		&aws_alb.DescribeTargetGroupsInput{
			Names:  awsNames,
			Marker: reqMarker,
		}).Return(&aws_alb.DescribeTargetGroupsOutput{NextMarker: nextMarker, TargetGroups: targetGroups}, err).Once()
}

func (m *mockALB) mockRegisterTargets(targetGroupARN, instanceID string, err error) {
	m.On("RegisterTargets", &aws_alb.RegisterTargetsInput{
		TargetGroupArn: aws.String(targetGroupARN),
		Targets:        []*aws_alb.TargetDescription{{Id: aws.String(instanceID)}},
	}).Return(&aws_alb.RegisterTargetsOutput{}, err)
}

func (m *mockALB) mockDeregisterTargets(targetGroupARN, instanceID string, err error) {
	m.On("DeregisterTargets", &aws_alb.DeregisterTargetsInput{
		TargetGroupArn: aws.String(targetGroupARN),
		Targets:        []*aws_alb.TargetDescription{{Id: aws.String(instanceID)}},
	}).Return(&aws_alb.DeregisterTargetsOutput{}, err)
}

func (m *mockMetadata) mockInstanceMetadata(instanceID string) {
	m.On("GetInstanceIdentityDocument").Return(ec2metadata.EC2InstanceIdentityDocument{InstanceID: instanceID}, nil)
}

func setup(targetGroupNames ...string) (controller.Updater, *mockALB, *mockMetadata) {
	a, _ := New(region, targetGroupNames, time.Nanosecond)
	mockALB := &mockALB{}
	mockMetadata := &mockMetadata{}
	a.(*alb).awsALB = mockALB
	a.(*alb).metadata = mockMetadata
	return a, mockALB, mockMetadata
}

func TestMetricsRegisteredCorrectly(t *testing.T) {
	//when
	_, _ = New(region, []string{"internal", "external"}, time.Nanosecond)

	//then
	assert.Equal(t, "feed_ingress_alb_frontends_attached", metricName(attachedFrontendGauge))
}

func metricName(c prometheus.Collector) string {
	descriptionCh := make(chan *prometheus.Desc, 1)
	c.Describe(descriptionCh)
	desc := <-descriptionCh
	descReflect := reflect.ValueOf(*desc)
	fqNameField := descReflect.FieldByName("fqName")
	return fqNameField.String()
}

func TestCanNotCreateUpdaterWithoutLabelValue(t *testing.T) {
	//when
	_, err := New(region, []string{}, time.Nanosecond)

	//then
	assert.Error(t, err)
}

func TestRegisterInstance(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"internal-arn", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("internal-arn", instanceID, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)

	//when
	err := a.Start()
	updateErr := a.Update(controller.IngressEntries{})

	//then
	mockALB.AssertExpectations(t)
	mockMetadata.AssertExpectations(t)
	assert.NoError(t, err)
	assert.NoError(t, updateErr)
	assert.NoError(t, a.Health())
}

func TestReportsErrorIfDidntRegisterAllTargetGroups(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"internal-arn", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("internal-arn", instanceID, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, errors.New("ka boom"))

	//when
	err := a.Start()
	updateErr := a.Update(controller.IngressEntries{})

	//then
	assert.NoError(t, err)
	assert.Error(t, updateErr)
}

func TestErrorGettingMetadata(t *testing.T) {
	e, _, mockMetadata := setup("group")
	mockMetadata.On("GetInstanceIdentityDocument").
		Return(ec2metadata.EC2InstanceIdentityDocument{}, errors.New("no metadata for you"))

	err := e.Update(controller.IngressEntries{})

	assert.Error(t, err)
}

func TestErrorDescribingTargetGroups(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("group")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"group"}, []string{"group-arn"}, nil, nil, errors.New("ba koom"))

	//when
	_ = a.Start()
	updateErr := a.Update(controller.IngressEntries{})

	//then
	assert.Error(t, updateErr)
}

func TestMissingTargetGroups(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)

	//when
	err := a.Start()
	updateErr := a.Update(controller.IngressEntries{})

	//then
	assert.NoError(t, err)
	assert.Error(t, updateErr)
}

func TestDescribeTargetGroupPages(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"internal-arn", ""},
		nil, aws.String("do-next"), nil)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"", "external-arn"},
		aws.String("do-next"), nil, nil)
	mockALB.mockRegisterTargets("internal-arn", instanceID, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)

	//when
	err := a.Start()
	updateErr := a.Update(controller.IngressEntries{})

	//then
	mockALB.AssertExpectations(t)
	mockMetadata.AssertExpectations(t)
	assert.NoError(t, err)
	assert.NoError(t, updateErr)
}

func TestDeregistersOnStop(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"internal-arn", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("internal-arn", instanceID, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)
	mockALB.mockDeregisterTargets("internal-arn", instanceID, nil)
	mockALB.mockDeregisterTargets("external-arn", instanceID, nil)

	//when
	_ = a.Start()
	_ = a.Update(controller.IngressEntries{})
	stopErr := a.Stop()

	//then
	mockALB.AssertExpectations(t)
	mockMetadata.AssertExpectations(t)
	assert.NoError(t, stopErr)
}

func TestDeregisterErrorIsHandledInStop(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"internal-arn", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("internal-arn", instanceID, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)
	mockALB.mockDeregisterTargets("internal-arn", instanceID, errors.New("ba boom"))
	mockALB.mockDeregisterTargets("external-arn", instanceID, nil)

	//when
	_ = a.Start()
	_ = a.Update(controller.IngressEntries{})
	stopErr := a.Stop()

	//then
	mockALB.AssertExpectations(t)
	mockMetadata.AssertExpectations(t)
	assert.NoError(t, stopErr)
}

func TestStopWaitsForDeregisterDelay(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"internal-arn", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("internal-arn", instanceID, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)
	mockALB.mockDeregisterTargets("internal-arn", instanceID, nil)
	mockALB.mockDeregisterTargets("external-arn", instanceID, nil)
	a.(*alb).targetGroupDeregistrationDelay = time.Millisecond * 50

	//when
	_ = a.Start()
	_ = a.Update(controller.IngressEntries{})
	beforeStop := time.Now()
	_ = a.Stop()
	stopDuration := time.Now().Sub(beforeStop)

	//then
	assert.True(t, stopDuration.Nanoseconds() > time.Millisecond.Nanoseconds()*50,
		"Drain time should have caused stop to take at least 50ms.")
}

func TestHealthReportsHealthyBeforeFirstUpdate(t *testing.T) {
	// given
	a, _, _ := setup("internal", "external")

	// when
	err := a.Start()

	// then
	assert.NoError(t, err)
	assert.Nil(t, a.Health())
}

func TestHealthReportsUnhealthyAfterUnsuccessfulFirstUpdate(t *testing.T) {
	//given
	a, mockALB, mockMetadata := setup("internal", "external")
	instanceID := "cow"
	mockMetadata.mockInstanceMetadata(instanceID)
	mockALB.mockDescribeTargetGroups([]string{"internal", "external"}, []string{"", "external-arn"},
		nil, nil, nil)
	mockALB.mockRegisterTargets("external-arn", instanceID, nil)

	//when
	err := a.Start()
	updateErr := a.Update(controller.IngressEntries{})

	//then
	assert.NoError(t, err)
	assert.Error(t, updateErr)
	assert.Error(t, a.Health())
}


<|start_filename|>util/cmd/csv_test.go<|end_filename|>
package cmd

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestSettingCSV(t *testing.T) {
	asserter := assert.New(t)

	csv := CommaSeparatedValues{}
	_ = csv.Set("first,second,third")

	asserter.Equal(CommaSeparatedValues{"first", "second", "third"}, csv)
}

func TestSettingEmptyCSV(t *testing.T) {
	asserter := assert.New(t)

	csv := CommaSeparatedValues{}
	_ = csv.Set("")

	asserter.Equal(CommaSeparatedValues{}, csv)
}


<|start_filename|>controller/updater.go<|end_filename|>
package controller

// Updater that the Controller delegates to.
type Updater interface {
	// Start the ingress updater, returning immediately after it's started.
	Start() error
	// Stop the ingress updater. Blocks until the ingress updater stops or an error occurs.
	Stop() error
	// Update the ingress updater configuration.
	// Not thread safe, should only be called by a single go routine
	Update(IngressEntries) error
	// Health returns nil if healthy, otherwise an error. Should be fast to respond, as it
	// may be called often. Any long running checks should be done separately.
	Health() error
}


<|start_filename|>merlin/mocks/merlin_client.go<|end_filename|>
// Package mocks generated by mockery v1.0.0
package mocks

import (
	"context"

	"github.com/golang/protobuf/ptypes/empty"
	"github.com/golang/protobuf/ptypes/wrappers"
	"github.com/sky-uk/merlin/types"
	"github.com/stretchr/testify/mock"
	"google.golang.org/grpc"
)

// MerlinClient is an autogenerated mock type for the MerlinClient type
type MerlinClient struct {
	mock.Mock
}

// CreateServer provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) CreateServer(ctx context.Context, in *types.RealServer, opts ...grpc.CallOption) (*empty.Empty, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *empty.Empty
	if rf, ok := ret.Get(0).(func(context.Context, *types.RealServer, ...grpc.CallOption) *empty.Empty); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*empty.Empty)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *types.RealServer, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// CreateService provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) CreateService(ctx context.Context, in *types.VirtualService, opts ...grpc.CallOption) (*empty.Empty, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *empty.Empty
	if rf, ok := ret.Get(0).(func(context.Context, *types.VirtualService, ...grpc.CallOption) *empty.Empty); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*empty.Empty)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *types.VirtualService, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// DeleteServer provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) DeleteServer(ctx context.Context, in *types.RealServer, opts ...grpc.CallOption) (*empty.Empty, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *empty.Empty
	if rf, ok := ret.Get(0).(func(context.Context, *types.RealServer, ...grpc.CallOption) *empty.Empty); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*empty.Empty)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *types.RealServer, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// DeleteService provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) DeleteService(ctx context.Context, in *wrappers.StringValue, opts ...grpc.CallOption) (*empty.Empty, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *empty.Empty
	if rf, ok := ret.Get(0).(func(context.Context, *wrappers.StringValue, ...grpc.CallOption) *empty.Empty); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*empty.Empty)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *wrappers.StringValue, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// List provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) List(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*types.ListResponse, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *types.ListResponse
	if rf, ok := ret.Get(0).(func(context.Context, *empty.Empty, ...grpc.CallOption) *types.ListResponse); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*types.ListResponse)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *empty.Empty, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// UpdateServer provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) UpdateServer(ctx context.Context, in *types.RealServer, opts ...grpc.CallOption) (*empty.Empty, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *empty.Empty
	if rf, ok := ret.Get(0).(func(context.Context, *types.RealServer, ...grpc.CallOption) *empty.Empty); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*empty.Empty)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *types.RealServer, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// UpdateService provides a mock function with given fields: ctx, in, opts
func (_m *MerlinClient) UpdateService(ctx context.Context, in *types.VirtualService, opts ...grpc.CallOption) (*empty.Empty, error) {
	_va := make([]interface{}, len(opts))
	for _i := range opts {
		_va[_i] = opts[_i]
	}
	var _ca []interface{}
	_ca = append(_ca, ctx, in)
	_ca = append(_ca, _va...)
	ret := _m.Called(_ca...)

	var r0 *empty.Empty
	if rf, ok := ret.Get(0).(func(context.Context, *types.VirtualService, ...grpc.CallOption) *empty.Empty); ok {
		r0 = rf(ctx, in, opts...)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*empty.Empty)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, *types.VirtualService, ...grpc.CallOption) error); ok {
		r1 = rf(ctx, in, opts...)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}


<|start_filename|>elb/elbstatus/status.go<|end_filename|>
/*
Package elbstatus provides an updater for an ELB frontend to update ingress statuses.
*/
package elbstatus

import (
	"fmt"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	awselb "github.com/aws/aws-sdk-go/service/elb"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/elb"
	"github.com/sky-uk/feed/k8s"
	k8sStatus "github.com/sky-uk/feed/k8s/status"
	v1 "k8s.io/api/core/v1"
)

// Config for creating a new ELB status updater.
type Config struct {
	Region              string
	FrontendTagValue    string
	IngressNameTagValue string
	KubernetesClient    k8s.Client
}

// New creates a new ELB frontend status updater.
func New(conf Config) (controller.Updater, error) {
	awsSession, err := session.NewSession(&aws.Config{Region: &conf.Region})
	if err != nil {
		return nil, fmt.Errorf("unable to create ELB status updater: %v", err)
	}

	return &status{
		awsElb:              awselb.New(awsSession),
		frontendTagValue:    conf.FrontendTagValue,
		ingressNameTagValue: conf.IngressNameTagValue,
		loadBalancers:       make(map[string]v1.LoadBalancerStatus),
		kubernetesClient:    conf.KubernetesClient,
	}, nil
}

type status struct {
	awsElb              elb.ELB
	frontendTagValue    string
	ingressNameTagValue string
	loadBalancers       map[string]v1.LoadBalancerStatus
	kubernetesClient    k8s.Client
}

// Start discovers the elbs and generates loadBalancer statuses.
func (s *status) Start() error {
	clusterFrontEnds, err := elb.FindFrontEndElbsWithIngressClassName(s.awsElb, s.frontendTagValue, s.ingressNameTagValue)
	if err != nil {
		return err
	}

	for lbLabel, clusterFrontEnd := range clusterFrontEnds {
		s.loadBalancers[lbLabel] = k8sStatus.GenerateLoadBalancerStatus([]string{clusterFrontEnd.DNSName})
	}
	return nil
}

func (s *status) Stop() error {
	return nil
}

func (s *status) Health() error {
	return nil
}

func (s *status) Update(ingresses controller.IngressEntries) error {
	return k8sStatus.Update(ingresses, s.loadBalancers, s.kubernetesClient)
}


<|start_filename|>util/cmd/cmd.go<|end_filename|>
package cmd

import (
	"fmt"
	"io"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"

	"github.com/onrik/logrus/filename"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/prometheus/client_golang/prometheus/push"

	log "github.com/sirupsen/logrus"
	"github.com/sky-uk/feed/util/metrics"
)

// Pulse represents something alive whose health can be checked.
type Pulse interface {
	// Health returns the current health, nil if healthy.
	Health() error
	// Stop the thing that's alive.
	Stop() error
}

// AddHealthPort is used to expose the health over http.
func AddHealthPort(pulse Pulse, healthPort int) {
	http.HandleFunc("/health", healthHandler(pulse))
	http.Handle("/metrics", promhttp.Handler())
	http.HandleFunc("/alive", okHandler)

	go func() {
		log.Error(http.ListenAndServe(":"+strconv.Itoa(healthPort), nil))
		log.Info(pulse.Stop())
		os.Exit(-1)
	}()
}

func healthHandler(pulse Pulse) func(w http.ResponseWriter, r *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		if err := pulse.Health(); err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			_, _ = io.WriteString(w, fmt.Sprintf("%v\n", err))
			return
		}

		w.WriteHeader(http.StatusOK)
		_, _ = io.WriteString(w, "ok\n")
	}
}

func okHandler(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
	_, _ = io.WriteString(w, "ok\n")
}

const pollInterval = time.Second

// AddUnhealthyLogger adds a periodic poller which reports an unhealthy status.
// The healthCounter is increased by pollInterval if unhealthy.
func addUnhealthyLogger(pulse Pulse, unhealthyCounter prometheus.Counter) {
	go func() {
		healthy := true
		tickCh := time.Tick(pollInterval)
		for range tickCh {
			if err := pulse.Health(); err != nil {
				unhealthyCounter.Add(pollInterval.Seconds())
				if healthy {
					log.Warnf("Unhealthy: %v", err)
					healthy = false
				}
			} else if !healthy {
				log.Info("Health restored")
				healthy = true
			}
		}
	}()
}

// AddSignalHandler allows the  controller to shutdown gracefully by respecting SIGTERM.
func AddSignalHandler(pulse Pulse) {
	c := make(chan os.Signal, 1)
	// SIGTERM is used by Kubernetes to gracefully stop pods.
	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		for sig := range c {
			log.Infof("Signalled %v, shutting down gracefully", sig)
			err := pulse.Stop()
			if err != nil {
				log.Errorf("Error while stopping: %v", err)
				os.Exit(-1)
			}
			os.Exit(0)
		}
	}()
}

// ConfigureLogging sets logging to Stdout and manages setting debug level
func ConfigureLogging(debug bool) {
	// logging is the main output, so write it all to stdout
	log.SetOutput(os.Stdout)
	if debug {
		log.SetLevel(log.DebugLevel)
	}
	filenameHook := filename.NewHook()
	filenameHook.Field = "source"
	log.AddHook(filenameHook)
}

// ConfigureMetrics sets up metrics pushing and default labels. This must be called before any metrics
// are defined.
func ConfigureMetrics(job string, prometheusLabels KeyValues, pushgatewayURL string, pushgatewayIntervalSeconds int) {
	labels := make(prometheus.Labels)
	for _, l := range prometheusLabels {
		labels[l.key] = l.value
	}
	metrics.SetConstLabels(labels)
	addMetricsPusher(job, pushgatewayURL, time.Second*time.Duration(pushgatewayIntervalSeconds))
}

// AddHealthMetrics adds a global health metric for the given pulse. This should only be called
// a single time per binary.
func AddHealthMetrics(pulse Pulse, prometheusSubsystem string) {
	unhealthyCounter := createUnhealthyCounter(prometheusSubsystem)
	addUnhealthyLogger(pulse, unhealthyCounter)
}

func createUnhealthyCounter(subsystem string) prometheus.Counter {
	return metrics.RegisterNewDefaultCounter(subsystem, "unhealthy_time",
		fmt.Sprintf("The number of seconds %s-%s has been unhealthy.",
			metrics.PrometheusNamespace, subsystem))
}

// AddMetricsPusher starts a periodic push of metrics to a prometheus pushgateway.
func addMetricsPusher(job, pushgatewayURL string, interval time.Duration) {
	if pushgatewayURL == "" {
		return
	}

	go func() {
		tick := time.Tick(interval)
		for range tick {
			instance, err := os.Hostname()
			if err != nil {
				log.Warnf("Unable to lookup hostname for metrics: %v", err)
				continue
			}

			if err := push.New(pushgatewayURL, job).Grouping("instance", instance).Push(); err != nil {
				log.Warnf("Unable to push metrics: %v", err)
			}
		}
	}()
}


<|start_filename|>gorb/gorb.go<|end_filename|>
/*
Package gorb provide the ability to register and deregister the backend
*/
package gorb

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"os/exec"
	"path"
	"strconv"
	"strings"
	"time"

	multierror "github.com/hashicorp/go-multierror"
	"github.com/sethgrid/pester"
	log "github.com/sirupsen/logrus"
	"github.com/sky-uk/feed/controller"
)

// PulseArgs defines health check URI
type PulseArgs struct {
	Method string `json:"method"`
	Path   string `json:"path"`
	Expect string `json:"expect"`
}

// Pulse defines backend health check
type Pulse struct {
	TypeHealthcheck string    `json:"type"`
	Args            PulseArgs `json:"args"`
	Interval        string    `json:"interval"`
}

// BackendConfig defines the backend configuration
type BackendConfig struct {
	Host   string `json:"host"`
	Port   int    `json:"port"`
	Method string `json:"method"`
	Weight int    `json:"weight"`
	Pulse  Pulse  `json:"pulse"`
}

// VirtualService defines the virtual services
type VirtualService struct {
	Name string
	Port int
}

// Config defines all the configuration required for gorb
type Config struct {
	ServerBaseURL              string
	InstanceIP                 string
	DrainDelay                 time.Duration
	ServicesDefinition         []VirtualService
	BackendWeight              int
	BackendMethod              string
	VipLoadbalancer            string
	ManageLoopback             bool
	BackendHealthcheckInterval string
	BackendHealthcheckType     string
	InterfaceProcFsPath        string
}

// Backend defines the backend configuration + service name
type backend struct {
	serviceName   string
	backendConfig BackendConfig
}

type loopbackAction string

const (
	addLoopback    loopbackAction = "add"
	deleteLoopback loopbackAction = "del"
)

// New creates a gorb handler
func New(c *Config) (controller.Updater, error) {
	if c.ServerBaseURL == "" {
		return nil, errors.New("unable to create Gorb updater: missing server ip address")
	}
	initMetrics()
	log.Infof("Gorb server url: %s, drainDelay: %v, instance ip adddress: %s, vipLoadbalancer: %s", c.ServerBaseURL, c.DrainDelay, c.InstanceIP, c.VipLoadbalancer)

	backendDefinitions := []backend{}

	var backendDefinition backend
	for _, service := range c.ServicesDefinition {
		pulse := Pulse{
			TypeHealthcheck: c.BackendHealthcheckType,
			Interval:        c.BackendHealthcheckInterval,
		}
		if c.BackendHealthcheckType == "http" {
			pulse.Args = PulseArgs{
				Method: "GET",
				Path:   "/",
				Expect: "404",
			}
		}

		backendDefinition = backend{
			serviceName: service.Name,
			backendConfig: BackendConfig{
				Host:   c.InstanceIP,
				Port:   service.Port,
				Method: c.BackendMethod,
				Weight: c.BackendWeight,
				Pulse:  pulse,
			},
		}

		backendDefinitions = append(backendDefinitions, backendDefinition)
	}

	httpClient := pester.New()
	httpClient.Timeout = time.Second * 2
	httpClient.MaxRetries = 3

	return &gorb{
		command:    &SimpleCommandRunner{},
		config:     c,
		backend:    backendDefinitions,
		httpClient: httpClient,
	}, nil
}

// CommandRunner is a cut down version of exec.Cmd for running commands
type CommandRunner interface {
	Execute(cmd string) ([]byte, error)
}

// SimpleCommandRunner implements CommandRunner
type SimpleCommandRunner struct {
}

// Execute runs the given command and returns its output
func (c *SimpleCommandRunner) Execute(cmd string) ([]byte, error) {
	log.Infof(fmt.Sprintf("Executing cmd: %s", cmd))
	return exec.Command("bash", "-c", cmd).Output()
}

type gorb struct {
	command    CommandRunner
	config     *Config
	httpClient *pester.Client
	backend    []backend
}

func (g *gorb) Start() error {
	return nil
}

// Stop removes this instance from Gorb
func (g *gorb) Stop() error {
	var errorArr *multierror.Error
	for _, backend := range g.backend {
		backend.backendConfig.Weight = 0
		err := g.modifyBackend(&backend)
		if err != nil {
			log.WithFields(log.Fields{"err": err}).Error("Unable to set the backend weight to 0")
			errorArr = multierror.Append(errorArr, err)
		} else {
			log.WithFields(log.Fields{"backend": backend}).Infof("Set backend weight to 0")
		}
	}

	log.Infof("Waiting %v to finish gorb draining", g.config.DrainDelay)
	time.Sleep(g.config.DrainDelay)

	if g.config.ManageLoopback {
		err := g.manageLoopBack(deleteLoopback)
		errorArr = multierror.Append(errorArr, err)
	}

	for _, backend := range g.backend {
		err := g.removeBackend(&backend)
		if err != nil {
			log.WithFields(log.Fields{"err": err}).Error("Unable to remove Backend ")
			errorArr = multierror.Append(errorArr, err)
		} else {
			log.WithFields(log.Fields{"backend": backend}).Infof("Backend succesfully removed")
		}
	}

	return errorArr.ErrorOrNil()
}

func (g *gorb) Health() error {
	resp, err := g.httpClient.Get(fmt.Sprintf("%s/service", g.config.ServerBaseURL))
	if err != nil {
		return fmt.Errorf("unable to check service details: %v", err)
	}

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("gorb server is not healthy. Status code: %d, Response Body %s", resp.StatusCode, resp.Body)
	}

	return nil
}

func (g *gorb) Update(controller.IngressEntries) error {
	var errorArr *multierror.Error
	if g.config.ManageLoopback {
		err := g.manageLoopBack(addLoopback)
		errorArr = multierror.Append(errorArr, err)
	}

	for _, backend := range g.backend {
		backendNotFound, err := g.backendNotFound(&backend)
		if err != nil {
			log.WithFields(log.Fields{"err": err, "backend": backend}).Error("Unable to check if backend already exists")
		}

		if backendNotFound {
			err := g.addBackend(&backend)
			if err != nil {
				log.WithFields(log.Fields{"err": err, "backend": backend}).Error("Error adding backend ")
				errorArr = multierror.Append(errorArr, err)
			} else {
				log.WithFields(log.Fields{"backend": backend}).Infof("Backend added successfully")
				attachedFrontendGauge.Set(float64(1))
			}
		}
	}

	return errorArr.ErrorOrNil()
}

func (g *gorb) manageLoopBack(action loopbackAction) error {
	var arpIgnore int
	var arpAnnounce int
	var expectedVipCount int
	var interfaceAction string

	switch action {
	// disable ARP for the loopback interface - see http://kb.linuxvirtualserver.org/wiki/Using_arp_announce/arp_ignore_to_disable_ARP
	case addLoopback:
		arpIgnore = 1
		arpAnnounce = 2
		interfaceAction = "add"
		expectedVipCount = 0
	case deleteLoopback:
		arpIgnore = 0
		arpAnnounce = 0
		interfaceAction = "del"
		expectedVipCount = 1
	default:
		return fmt.Errorf("unsupported loopback action %s", action)
	}

	var errorArr *multierror.Error
	vipLoadbalancers := strings.Split(g.config.VipLoadbalancer, ",")
	for index, vip := range vipLoadbalancers {
		vipCount, err := g.loopbackInterfaceCount(fmt.Sprintf("lo:%d", index), vip)
		errorArr = multierror.Append(errorArr, err)
		if vipCount == expectedVipCount {
			_, err = g.command.Execute(fmt.Sprintf("sudo ip addr %s %s/32 dev lo label lo:%d", interfaceAction, vip, index))
			errorArr = multierror.Append(errorArr, err)
		}

		_, err = g.command.Execute(fmt.Sprintf("echo %d | sudo tee %s > /dev/null", arpIgnore, path.Join(g.config.InterfaceProcFsPath, "arp_ignore")))
		errorArr = multierror.Append(errorArr, err)

		_, err = g.command.Execute(fmt.Sprintf("echo %d | sudo tee %s > /dev/null", arpAnnounce, path.Join(g.config.InterfaceProcFsPath, "arp_announce")))
		errorArr = multierror.Append(errorArr, err)
	}

	return errorArr.ErrorOrNil()
}

func (g *gorb) loopbackInterfaceCount(label string, vip string) (int, error) {
	cmdOutput, err := g.command.Execute(fmt.Sprintf("sudo ip addr show label %s | grep -c %s/32 | xargs echo", label, vip))
	if err != nil {
		return -1, fmt.Errorf("unable to check whether loopback interface exists for label: %s and vip: %s, error %v", label, vip, err)
	}

	vipCount, err := strconv.Atoi(strings.TrimSpace(string(cmdOutput)))
	if err != nil {
		return -1, fmt.Errorf("unable to parse loopback interface count from the output: %s, error :%v", string(cmdOutput), err)
	}
	return vipCount, nil
}

func (g *gorb) backendNotFound(backend *backend) (bool, error) {
	resp, err := g.httpClient.Get(g.serviceRequest(backend))
	if err != nil {
		return false, fmt.Errorf("unable to retrieve backend details for instance ip: %s, error :%v", g.config.InstanceIP, err)
	}
	defer resp.Body.Close()
	return resp.StatusCode == http.StatusNotFound, nil
}

func (g *gorb) addBackend(backend *backend) error {
	payload, err := json.Marshal(backend.backendConfig)
	if err != nil {
		return fmt.Errorf("error while marshalling backend: %v, error: %v", backend, err)
	}

	err = g.executeRequest("PUT", backend, bytes.NewReader(payload))
	if err != nil {
		return fmt.Errorf("error while adding backend: %v, error: %v", backend, err)
	}
	return nil
}

func (g *gorb) modifyBackend(backend *backend) error {
	payload, err := json.Marshal(backend.backendConfig)
	if err != nil {
		return fmt.Errorf("error while marshalling backend: %v, error: %v", backend, err)
	}

	err = g.executeRequest("PATCH", backend, bytes.NewReader(payload))
	if err != nil {
		return fmt.Errorf("error while modifying backend: %v, error: %v", backend, err)
	}
	return nil
}

func (g *gorb) removeBackend(backend *backend) error {
	err := g.executeRequest("DELETE", backend, nil)
	if err != nil {
		return fmt.Errorf("error while removing backend: %v, error: %v", backend, err)
	}
	return nil
}

func (g *gorb) executeRequest(method string, backend *backend, payload io.Reader) error {
	req, err := http.NewRequest(method, g.serviceRequest(backend), payload)
	if err != nil {
		return fmt.Errorf("error while creating %s request: %v", method, err)
	}

	resp, err := g.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("error while sending %s backend request, error: %v", method, err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		log.WithFields(log.Fields{"StatusCode": resp.StatusCode}).Infof("Error while sending %s backend request", method)
		return fmt.Errorf("failed to send %s backend request, status code: %d, response: %v", method, resp.StatusCode, body)
	}
	return nil
}

func (g *gorb) serviceRequest(backend *backend) string {
	return fmt.Sprintf("%s/service/%s/node-%s-%s", g.config.ServerBaseURL, backend.serviceName, backend.serviceName, g.config.InstanceIP)
}

func (g *gorb) String() string {
	return "Gorb frontend"
}


<|start_filename|>dns/r53/client.go<|end_filename|>
package r53

import (
	"fmt"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/route53"
	"github.com/sky-uk/feed/util"
)

const maxRecordChanges = 100

// Route53Client is the public interface
type Route53Client interface {
	GetHostedZoneDomain() (string, error)
	UpdateRecordSets(changes []*route53.Change) error
	GetRecords() ([]*route53.ResourceRecordSet, error)
}

// r53 interface exposes the subset of methods we use of the aws sdk
type r53 interface {
	GetHostedZone(input *route53.GetHostedZoneInput) (*route53.GetHostedZoneOutput, error)
	ChangeResourceRecordSets(input *route53.ChangeResourceRecordSetsInput) (*route53.ChangeResourceRecordSetsOutput, error)
	ListResourceRecordSets(input *route53.ListResourceRecordSetsInput) (*route53.ListResourceRecordSetsOutput, error)
}

// Route53Client enables interaction with aws route53
type client struct {
	r53              r53
	hostedZone       string
	maxRecordChanges int
}

// New creates a route53 client used to interact with aws
func New(hostedZone string, retries int) Route53Client {
	config := aws.Config{MaxRetries: aws.Int(retries)}
	awsSession, _ := session.NewSession()
	return &client{
		r53:              route53.New(awsSession, &config),
		hostedZone:       hostedZone,
		maxRecordChanges: maxRecordChanges,
	}
}

// GetHostedZoneDomain gets the domain for the hosted zone
func (dns *client) GetHostedZoneDomain() (string, error) {
	input := &route53.GetHostedZoneInput{Id: aws.String(dns.hostedZone)}
	hostedZone, err := dns.r53.GetHostedZone(input)
	if err != nil {
		return "", fmt.Errorf("unable to get Hosted Zone Info: %v", err)
	}
	return *hostedZone.HostedZone.Name, nil
}

// UpdateRecordSets updates records in aws based on the change list.
func (dns *client) UpdateRecordSets(changes []*route53.Change) error {
	partitions := util.Partition(len(changes), dns.maxRecordChanges)
	for _, partition := range partitions {
		batch := changes[partition.Low:partition.High]
		recordSetsInput := &route53.ChangeResourceRecordSetsInput{
			HostedZoneId: aws.String(dns.hostedZone),
			ChangeBatch: &route53.ChangeBatch{
				Changes: batch,
			},
		}

		_, err := dns.r53.ChangeResourceRecordSets(recordSetsInput)

		if err != nil {
			return fmt.Errorf("failed to create A record: %v", err)
		}
	}

	return nil
}

// GetRecords gets a list of DNS records from aws.
func (dns *client) GetRecords() ([]*route53.ResourceRecordSet, error) {
	var records []*route53.ResourceRecordSet
	request := &route53.ListResourceRecordSetsInput{
		HostedZoneId: aws.String(dns.hostedZone),
	}
	for {
		recordSetsOutput, err := dns.r53.ListResourceRecordSets(request)

		if err != nil {
			return nil, fmt.Errorf("failed to fetch A records: %v", err)
		}

		recordSets := recordSetsOutput.ResourceRecordSets

		for _, recordSet := range recordSets {
			if *recordSet.Type == route53.RRTypeA || *recordSet.Type == route53.RRTypeCname {
				records = append(records, recordSet)
			}
		}

		if !aws.BoolValue(recordSetsOutput.IsTruncated) {
			break
		}

		request = &route53.ListResourceRecordSetsInput{
			HostedZoneId:    aws.String(dns.hostedZone),
			StartRecordName: recordSetsOutput.NextRecordName,
			StartRecordType: recordSetsOutput.NextRecordType,
		}
	}

	return records, nil
}


<|start_filename|>feed-ingress/cmd/nlb.go<|end_filename|>
package cmd

import (
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/elb"
	"github.com/sky-uk/feed/k8s"
	"github.com/sky-uk/feed/nlb"
	"github.com/sky-uk/feed/nlb/nlbstatus"
	"github.com/spf13/cobra"
)

var nlbCmd = &cobra.Command{
	Use:   "nlb",
	Short: "Attach to AWS Network Load Balancers",
	Run: func(cmd *cobra.Command, args []string) {
		runCmd(appendNlbIngressUpdaters)
	},
}

func init() {
	rootCmd.AddCommand(nlbCmd)

	nlbCmd.Flags().StringVar(&region, "region", defaultRegion,
		"AWS region for frontend attachment.")
	nlbCmd.Flags().StringVar(&elbFrontendTagValue, "nlb-frontend-tag-value", defaultLbFrontendTagValue,
		"Attach to NLBs tagged with "+elb.FrontendTag+"=value. Leave empty to not attach.")
	nlbCmd.Flags().IntVar(&elbExpectedNumber, "nlb-expected-number", defaultLbExpectedNumber,
		"Expected number of NLBs to attach to. If 0 the controller will not check,"+
			" otherwise it fails to start if it can't attach to this number.")
	nlbCmd.Flags().DurationVar(&drainDelay, "drain-delay", defaultDrainDelay, "Delay to wait"+
		" for feed-ingress to drain from the registration component on shutdown. Should match the NLB's drain time.")
}

func appendNlbIngressUpdaters(kubernetesClient k8s.Client, updaters []controller.Updater) ([]controller.Updater, error) {
	updater, err := nlb.New(region, elbFrontendTagValue, ingressClassName, elbExpectedNumber, drainDelay)
	if err != nil {
		return nil, err
	}
	updaters = append(updaters, updater)

	statusConfig := nlbstatus.Config{
		Region:              region,
		FrontendTagValue:    elbFrontendTagValue,
		IngressNameTagValue: ingressClassName,
		KubernetesClient:    kubernetesClient,
	}
	statusUpdater, err := nlbstatus.New(statusConfig)
	if err != nil {
		return nil, err
	}
	return append(updaters, statusUpdater), nil
}


<|start_filename|>feed-ingress/cmd/elb.go<|end_filename|>
package cmd

import (
	"time"

	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/elb"
	elbstatus "github.com/sky-uk/feed/elb/elbstatus"
	"github.com/sky-uk/feed/k8s"
	"github.com/spf13/cobra"
)

var (
	region                         string
	elbFrontendTagValue            string
	elbExpectedNumber              int
	drainDelay                     time.Duration
	targetGroupNames               []string
	targetGroupDeregistrationDelay time.Duration
)

const (
	defaultRegion                         = "eu-west-1"
	defaultLbFrontendTagValue             = ""
	defaultLbExpectedNumber               = 0
	defaultDrainDelay                     = time.Second * 60
	defaultTargetGroupDeregistrationDelay = time.Second * 300
)

var elbCmd = &cobra.Command{
	Use:   "elb",
	Short: "Attach to AWS Classic Load Balancers",
	Run: func(cmd *cobra.Command, args []string) {
		runCmd(appendElbIngressUpdaters)
	},
}

func init() {
	rootCmd.AddCommand(elbCmd)

	elbCmd.Flags().StringVar(&region, "region", defaultRegion,
		"AWS region for frontend attachment.")
	elbCmd.Flags().StringVar(&elbFrontendTagValue, "elb-frontend-tag-value", defaultLbFrontendTagValue,
		"Attach to ELBs tagged with "+elb.FrontendTag+"=value. Leave empty to not attach.")
	elbCmd.Flags().IntVar(&elbExpectedNumber, "elb-expected-number", defaultLbExpectedNumber,
		"Expected number of ELBs to attach to. If 0 the controller will not check,"+
			" otherwise it fails to start if it can't attach to this number.")
	elbCmd.Flags().DurationVar(&drainDelay, "drain-delay", defaultDrainDelay, "Delay to wait"+
		" for feed-ingress to drain from the registration component on shutdown. Should match the ELB's drain time.")
}

func appendElbIngressUpdaters(kubernetesClient k8s.Client, updaters []controller.Updater) ([]controller.Updater, error) {
	elbUpdater, err := elb.New(region, elbFrontendTagValue, ingressClassName, elbExpectedNumber, drainDelay)
	if err != nil {
		return nil, err
	}
	updaters = append(updaters, elbUpdater)

	statusConfig := elbstatus.Config{
		Region:              region,
		FrontendTagValue:    elbFrontendTagValue,
		IngressNameTagValue: ingressClassName,
		KubernetesClient:    kubernetesClient,
	}
	elbStatusUpdater, err := elbstatus.New(statusConfig)
	if err != nil {
		return nil, err
	}
	return append(updaters, elbStatusUpdater), nil
}


<|start_filename|>util/slices.go<|end_filename|>
package util

// Range represents part of a slice
// High and low can be used with a golang slice [Low:High]
type Range struct {
	// Low is inclusive
	Low int
	// High is exclusive
	High int
}

// Partition calculates the indexes for splitting a slice of size length
// with a max partition size of size
func Partition(length int, size int) []Range {
	var result []Range
	for i := 0; i < length; i += size {
		upperBound := min(i+size, length)
		result = append(result, Range{Low: i, High: upperBound})
	}
	return result
}

func min(x int, y int) int {
	if x < y {
		return x
	}
	return y
}


<|start_filename|>gorb/gorb_metrics.go<|end_filename|>
package gorb

import (
	"sync"

	"github.com/sky-uk/feed/util/metrics"

	"github.com/prometheus/client_golang/prometheus"
)

var once sync.Once
var attachedFrontendGauge prometheus.Gauge

func initMetrics() {
	once.Do(func() {
		attachedFrontendGauge = metrics.RegisterNewDefaultGauge(metrics.PrometheusIngressSubsystem,
			"gorb_frontends_attached", "The total number of frontends attached to Gorb")
	})
}


<|start_filename|>merlin/merlin_test.go<|end_filename|>
package merlin

import (
	"errors"
	"fmt"
	"testing"
	"time"

	"github.com/gogo/protobuf/proto"
	"github.com/golang/protobuf/ptypes"
	"github.com/golang/protobuf/ptypes/empty"
	"github.com/golang/protobuf/ptypes/wrappers"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/merlin/mocks"
	"github.com/sky-uk/merlin/types"
	"github.com/stretchr/testify/mock"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func TestMerlin(t *testing.T) {
	RegisterFailHandler(Fail)
	RunSpecs(t, "Merlin Updater Test Suite")
}

var _ = Describe("merlin", func() {
	var (
		merlin        controller.Updater
		client        *mocks.MerlinClient
		nlMock        *netlinkMockType
		closeableMock *closeableMockType
		conf          Config

		healthCheck   *types.RealServer_HealthCheck
		emptyResponse = &empty.Empty{}

		expectedServer      *types.RealServer
		expectedHTTPSServer *types.RealServer
	)

	BeforeEach(func() {
		client = &mocks.MerlinClient{}
		nlMock = &netlinkMockType{}
		closeableMock = &closeableMockType{}
		closeableMock.On("Close").Return(nil)
		conf = Config{
			ServiceID:           "service1",
			HTTPSServiceID:      "https-service1",
			InstanceIP:          "172.16.16.1",
			InstancePort:        uint16(8080),
			InstanceHTTPSPort:   uint16(8443),
			ForwardMethod:       "route",
			HealthPort:          uint16(8081),
			HealthPath:          "health",
			HealthUpThreshold:   uint32(4),
			HealthDownThreshold: uint32(2),
			HealthPeriod:        time.Second * 10,
			HealthTimeout:       time.Second,
			DrainDelay:          time.Millisecond,
		}
	})

	JustBeforeEach(func() {
		m, err := New(conf)
		Expect(err).ToNot(HaveOccurred())
		m.(*updater).clientFactory = func(c *Config) (types.MerlinClient, closeable, error) {
			return client, closeableMock, nil
		}
		m.(*updater).nl = nlMock
		merlin = m
		healthCheck = &types.RealServer_HealthCheck{
			Endpoint:      &wrappers.StringValue{Value: fmt.Sprintf("http://:%d/%s", conf.HealthPort, conf.HealthPath)},
			UpThreshold:   conf.HealthUpThreshold,
			DownThreshold: conf.HealthDownThreshold,
			Period:        ptypes.DurationProto(conf.HealthPeriod),
			Timeout:       ptypes.DurationProto(conf.HealthTimeout),
		}

		expectedServer = &types.RealServer{
			ServiceID: conf.ServiceID,
			Key: &types.RealServer_Key{
				Ip:   conf.InstanceIP,
				Port: uint32(conf.InstancePort),
			},
			Config: &types.RealServer_Config{
				Weight:  &wrappers.UInt32Value{Value: 1},
				Forward: types.ForwardMethod_ROUTE,
			},
			HealthCheck: healthCheck,
		}
		expectedHTTPSServer = proto.Clone(expectedServer).(*types.RealServer)
		expectedHTTPSServer.ServiceID = conf.HTTPSServiceID
		expectedHTTPSServer.Key.Port = uint32(conf.InstanceHTTPSPort)
	})

	It("registers itself on start", func() {
		client.On("CreateServer", mock.Anything, expectedServer).Return(emptyResponse, nil)
		client.On("CreateServer", mock.Anything, expectedHTTPSServer).Return(emptyResponse, nil)

		err := merlin.Start()

		Expect(err).ToNot(HaveOccurred())
		client.AssertExpectations(GinkgoT())
		closeableMock.AssertExpectations(GinkgoT())
	})

	It("updates itself on start if already exists", func() {
		// http
		client.On("CreateServer", mock.Anything, expectedServer).Return(emptyResponse,
			status.Error(codes.AlreadyExists, "already exists"))
		client.On("UpdateServer", mock.Anything, expectedServer).Return(emptyResponse, nil)
		// https
		client.On("CreateServer", mock.Anything, expectedHTTPSServer).Return(emptyResponse,
			status.Error(codes.AlreadyExists, "already exists"))
		client.On("UpdateServer", mock.Anything, expectedHTTPSServer).Return(emptyResponse, nil)

		err := merlin.Start()

		Expect(err).ToNot(HaveOccurred())
		client.AssertExpectations(GinkgoT())
	})

	It("deregisters itself on stop", func() {
		// http
		drainServer := expectedServer
		drainServer.Config = &types.RealServer_Config{Weight: &wrappers.UInt32Value{Value: 0}}
		drainServer.HealthCheck = nil
		delServer := proto.Clone(drainServer).(*types.RealServer)
		delServer.Config = nil
		// https
		drainHTTPSServer := expectedHTTPSServer
		drainHTTPSServer.Config = &types.RealServer_Config{Weight: &wrappers.UInt32Value{Value: 0}}
		drainHTTPSServer.HealthCheck = nil
		delHTTPSServer := proto.Clone(drainHTTPSServer).(*types.RealServer)
		delHTTPSServer.Config = nil

		client.On("UpdateServer", mock.Anything, drainServer).Return(emptyResponse, nil)
		client.On("DeleteServer", mock.Anything, delServer).Return(emptyResponse, nil)
		client.On("UpdateServer", mock.Anything, drainHTTPSServer).Return(emptyResponse, nil)
		client.On("DeleteServer", mock.Anything, delHTTPSServer).Return(emptyResponse, nil)

		err := merlin.Stop()

		Expect(err).ToNot(HaveOccurred())
		client.AssertExpectations(GinkgoT())
		closeableMock.AssertExpectations(GinkgoT())
	})

	Context("service IDs are empty", func() {
		BeforeEach(func() {
			conf.ServiceID = ""
			conf.HTTPSServiceID = ""
		})

		It("doesn't register any servers", func() {
			err := merlin.Start()

			Expect(err).ToNot(HaveOccurred())
			client.AssertExpectations(GinkgoT())
		})

		It("doesn't deregister any servers", func() {
			err := merlin.Stop()

			Expect(err).ToNot(HaveOccurred())
			client.AssertExpectations(GinkgoT())
		})
	})

	Context("instance port is unset", func() {
		BeforeEach(func() {
			conf.InstancePort = 0
			conf.InstanceHTTPSPort = 0
		})

		It("doesn't register any servers", func() {
			err := merlin.Start()

			Expect(err).ToNot(HaveOccurred())
			client.AssertExpectations(GinkgoT())
		})

		It("doesn't deregister any servers", func() {
			err := merlin.Stop()

			Expect(err).ToNot(HaveOccurred())
			client.AssertExpectations(GinkgoT())
		})
	})

	Context("manages VIP", func() {
		BeforeEach(func() {
			conf.VIPInterface = "eth1"
			conf.VIP = "10.10.10.1"
		})

		It("should add VIP on start", func() {
			nlMock.On("addVIP", conf.VIPInterface, conf.VIP).Return(nil)
			client.On("CreateServer", mock.Anything, mock.Anything).Return(emptyResponse, nil)

			err := merlin.Start()

			Expect(err).ToNot(HaveOccurred())
			nlMock.AssertExpectations(GinkgoT())
		})

		It("should remove VIP on stop", func() {
			nlMock.On("removeVIP", conf.VIPInterface, conf.VIP).Return(nil)
			client.On("UpdateServer", mock.Anything, mock.Anything).Return(emptyResponse, nil)
			client.On("DeleteServer", mock.Anything, mock.Anything).Return(emptyResponse, nil)

			err := merlin.Stop()

			Expect(err).ToNot(HaveOccurred())
			nlMock.AssertExpectations(GinkgoT())
		})

		It("should remove the VIP if attach fails on start", func() {
			nlMock.On("addVIP", conf.VIPInterface, conf.VIP).Return(nil)
			client.On("CreateServer", mock.Anything, mock.Anything).Return(nil, errors.New("kaboom"))
			nlMock.On("removeVIP", conf.VIPInterface, conf.VIP).Return(nil)

			err := merlin.Start()

			Expect(err).To(HaveOccurred())
			nlMock.AssertExpectations(GinkgoT())
		})
	})
})

type netlinkMockType struct {
	mock.Mock
}

func (m *netlinkMockType) addVIP(vipInterface, vip string) error {
	args := m.Called(vipInterface, vip)
	return args.Error(0)
}

func (m *netlinkMockType) removeVIP(vipInterface, vip string) error {
	args := m.Called(vipInterface, vip)
	return args.Error(0)
}

type closeableMockType struct {
	mock.Mock
}

func (m *closeableMockType) Close() error {
	args := m.Called()
	return args.Error(0)
}


<|start_filename|>util/safe.go<|end_filename|>
package util

import (
	"sync"
)

// SafeBool is a thread safe boolean
type SafeBool struct {
	val bool
	sync.Mutex
}

// Get the value inside the SafeBool
func (s *SafeBool) Get() bool {
	s.Lock()
	defer s.Unlock()
	return s.val
}

// Set the value inside the SafeBool
func (s *SafeBool) Set(newVal bool) {
	s.Lock()
	s.val = newVal
	s.Unlock()
}

// SafeError is a thread safe error
type SafeError struct {
	val error
	sync.Mutex
}

// Get the value inside the SafeError
func (s *SafeError) Get() error {
	s.Lock()
	defer s.Unlock()
	return s.val
}

// Set the value inside the SafeError
func (s *SafeError) Set(newVal error) {
	s.Lock()
	s.val = newVal
	s.Unlock()
}

// SafeInt is a thread safe int
type SafeInt struct {
	val int
	sync.Mutex
}

// Get the value inside SafeInt
func (s *SafeInt) Get() int {
	s.Lock()
	defer s.Unlock()
	return s.val
}

// Add a value to the SafeInt
func (s *SafeInt) Add(addend int) int {
	s.Lock()
	defer s.Unlock()
	s.val += addend
	return s.val
}

// Set the value of the SafeInt
func (s *SafeInt) Set(val int) int {
	s.Lock()
	defer s.Unlock()
	s.val = val
	return s.val
}


<|start_filename|>k8s/watcher.go<|end_filename|>
package k8s

import (
	"sync"
	"time"

	log "github.com/sirupsen/logrus"
)

// Watcher provides channels for receiving updates. It tries its best to run forever, retrying
// if the underlying connection fails.
type Watcher interface {
	// Updates provides update notification.
	Updates() <-chan interface{}
}

type baseWatcher struct {
	updates chan interface{}
}

type watcher struct {
	baseWatcher
	sync.Mutex
}

func newBaseWatcher() baseWatcher {
	return baseWatcher{
		updates: make(chan interface{}, 1),
	}
}

// NewWatcher returns an initialized watcher.
func newWatcher() *watcher {
	return &watcher{baseWatcher: newBaseWatcher()}
}

func (w *watcher) Updates() <-chan interface{} {
	return w.updates
}

type bufferedWatcher struct {
	*watcher
	shouldUpdate bool
	sync.Mutex
}

func newBufferedWatcher(bufferTime time.Duration) *bufferedWatcher {
	b := &bufferedWatcher{watcher: newWatcher()}

	go func() {
		tick := time.Tick(bufferTime)
		for range tick {
			b.sendUpdate()
		}
	}()

	return b
}

func (b *bufferedWatcher) bufferUpdate() {
	b.Lock()
	defer b.Unlock()
	b.shouldUpdate = true
}

func (b *bufferedWatcher) sendUpdate() {
	b.Lock()
	defer b.Unlock()
	if b.shouldUpdate {
		b.shouldUpdate = false
		go func() { b.updates <- struct{}{} }()
	}
}

type combinedWatcher struct {
	baseWatcher
	watchers []Watcher
}

// CombineWatchers returns a watcher that watches all. The combined watcher becomes the owner
// of the passed in watchers and so clients should not attempt to use or stop the individual watchers.
func CombineWatchers(watchers ...Watcher) Watcher {
	combined := &combinedWatcher{baseWatcher: newBaseWatcher(), watchers: watchers}

	combiner := func(w Watcher) {
		for {
			select {
			case update := <-w.Updates():
				if update == nil {
					log.Panic("update should not be nil, did you close the watcher?")
				}
				combined.updates <- update
			}
		}
	}

	for _, w := range watchers {
		go combiner(w)
	}

	return combined
}

func (w *combinedWatcher) Updates() <-chan interface{} {
	return w.updates
}


<|start_filename|>feed-ingress/cmd/alb.go<|end_filename|>
package cmd

import (
	"github.com/sky-uk/feed/alb"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/k8s"

	"github.com/spf13/cobra"
)

var albCmd = &cobra.Command{
	Use:   "alb",
	Short: "Attach to AWS Application Load Balancers",
	Long: `Unfortunately, ALBs have a bug that prevents non-disruptive deployments of feed.
Specifically, they don't respect the deregistration delay. As a result, we don't recommend using ALBs at this time.`,
	Run: func(cmd *cobra.Command, args []string) {
		runCmd(appendAlbIngressUpdaters)
	},
}

func init() {
	rootCmd.AddCommand(albCmd)

	albCmd.Flags().StringVar(&region, "region", defaultRegion,
		"AWS region for frontend attachment.")
	albCmd.Flags().StringSliceVar(&targetGroupNames, "alb-target-group-names", []string{},
		"Names of ALB target groups to attach to, separated by commas.")
	albCmd.Flags().DurationVar(&targetGroupDeregistrationDelay, "alb-target-group-deregistration-delay",
		defaultTargetGroupDeregistrationDelay,
		"Delay to wait for feed-ingress to deregister from the ALB target group on shutdown. Should match"+
			" the target group setting in AWS.")
}

func appendAlbIngressUpdaters(kubernetesClient k8s.Client, updaters []controller.Updater) ([]controller.Updater, error) {
	albUpdater, err := alb.New(region, targetGroupNames, targetGroupDeregistrationDelay)
	if err != nil {
		return nil, err
	}
	return append(updaters, albUpdater), nil
}


<|start_filename|>dns/adapter/static_hostname_adapter.go<|end_filename|>
package adapter

import (
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/route53"
)

type staticHostnameAdapter struct {
	addressesWithScheme map[string]string
	ttl                 *int64
}

// NewStaticHostnameAdapter creates a FrontendAdapter which interacts with load balancers accessed by static hostnames.
func NewStaticHostnameAdapter(addressesWithScheme map[string]string, ttl time.Duration) FrontendAdapter {
	return &staticHostnameAdapter{addressesWithScheme, aws.Int64(int64(ttl.Seconds()))}
}

func (s *staticHostnameAdapter) Initialise() (map[string]DNSDetails, error) {
	schemeToFrontendMap := make(map[string]DNSDetails)
	for scheme, address := range s.addressesWithScheme {
		schemeToFrontendMap[scheme] = DNSDetails{DNSName: address}
	}

	return schemeToFrontendMap, nil
}

func (s *staticHostnameAdapter) CreateChange(action string, host string, details DNSDetails,
	recordExists bool, existingRecord *ConsolidatedRecord) *route53.Change {

	if recordExists && existingRecord.TTL != *s.ttl || !recordExists || action == "DELETE" {
		rrs := &route53.ResourceRecordSet{
			Name: aws.String(host),
			Type: aws.String("CNAME"),
			TTL:  s.ttl,
			ResourceRecords: []*route53.ResourceRecord{
				{
					Value: aws.String(details.DNSName),
				},
			},
		}

		return &route53.Change{
			Action:            aws.String(action),
			ResourceRecordSet: rrs,
		}
	}

	return nil
}

func (s *staticHostnameAdapter) IsManaged(rrs *route53.ResourceRecordSet) (*ConsolidatedRecord, bool) {
	if *rrs.Type == route53.RRTypeCname {
		record := ConsolidatedRecord{
			Name:     *rrs.Name,
			PointsTo: *rrs.ResourceRecords[0].Value,
		}
		if rrs.TTL != nil {
			record.TTL = *rrs.TTL
		}
		return &record, true
	}

	return nil, false
}


<|start_filename|>util/metrics/prometheus.go<|end_filename|>
package metrics

import (
	"sync"

	"github.com/prometheus/client_golang/prometheus"
	log "github.com/sirupsen/logrus"
)

const (
	// PrometheusNamespace is the metric namespace for feed binaries.
	PrometheusNamespace = "feed"
	// PrometheusIngressSubsystem is the metric subsystem for feed-ingress.
	PrometheusIngressSubsystem = "ingress"
	// PrometheusDNSSubsystem is the metric subsystem for feed-dns.
	PrometheusDNSSubsystem = "dns"
)

var labelsLock sync.Mutex
var constLabels prometheus.Labels

// ConstLabels should be used when creating a prometheus metric, as a set of default labels.
// To ensure the correct const labels are used, make sure metrics are not created in init().
// SetConstLabels should have been called first.
func ConstLabels() prometheus.Labels {
	labelsLock.Lock()
	defer labelsLock.Unlock()

	if constLabels == nil {
		log.Panic("Bug: ConstLabels() hasn't been set yet. Are you initialising metrics too early?")
	}
	return constLabels
}

// SetConstLabels sets default labels for prometheus metrics, that can be retrieved using
// ConstLabels().
func SetConstLabels(l prometheus.Labels) {
	labelsLock.Lock()
	defer labelsLock.Unlock()

	if constLabels != nil {
		log.Panic("Bug: ConstLabels() were already set, tried to set them multiple times.")
	}
	constLabels = l
}

func gaugeOpts(subsystem, name, help string) prometheus.GaugeOpts {
	return prometheus.GaugeOpts{
		Namespace:   PrometheusNamespace,
		Subsystem:   subsystem,
		Name:        name,
		Help:        help,
		ConstLabels: ConstLabels(),
	}
}

func counterOpts(subsystem, name, help string) prometheus.CounterOpts {
	return prometheus.CounterOpts{
		Namespace:   PrometheusNamespace,
		Subsystem:   subsystem,
		Name:        name,
		Help:        help,
		ConstLabels: ConstLabels(),
	}
}

func register(collector prometheus.Collector, name string) prometheus.Collector {
	err := prometheus.Register(collector)
	if err != nil {
		log.Fatalf("Could not register collector %s: %v", name, err)
	}
	return collector
}

// RegisterNewDefaultGauge creates and registers a named Gauge with default options
func RegisterNewDefaultGauge(subsystem, name, help string) prometheus.Gauge {
	return register(prometheus.NewGauge(gaugeOpts(subsystem, name, help)), name).(prometheus.Gauge)
}

// RegisterNewDefaultGaugeVec creates and registers a named GaugeVec with default options
func RegisterNewDefaultGaugeVec(subsystem, name, help string, labelNames []string) *prometheus.GaugeVec {
	return register(prometheus.NewGaugeVec(gaugeOpts(subsystem, name, help), labelNames), name).(*prometheus.GaugeVec)
}

// RegisterNewDefaultCounter creates and registers a named Counter with default options
func RegisterNewDefaultCounter(subsystem, name, help string) prometheus.Counter {
	return register(prometheus.NewCounter(counterOpts(subsystem, name, help)), name).(prometheus.Counter)
}


<|start_filename|>dns/dns_metrics.go<|end_filename|>
package dns

import (
	"sync"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/sky-uk/feed/util/metrics"
)

var once sync.Once
var recordsGauge prometheus.Gauge
var updateCount, failedCount, skippedCount prometheus.Counter

func initMetrics() {
	once.Do(func() {
		recordsGauge = metrics.RegisterNewDefaultGauge(metrics.PrometheusDNSSubsystem,
			"route53_records", "The current number of records.")
		updateCount = metrics.RegisterNewDefaultCounter(metrics.PrometheusDNSSubsystem,
			"route53_updates", "The number of record updates to Route53.")
		failedCount = metrics.RegisterNewDefaultCounter(metrics.PrometheusDNSSubsystem,
			"route53_failures", "The number of failed updates to route53.")
		skippedCount = metrics.RegisterNewDefaultCounter(metrics.PrometheusDNSSubsystem,
			"skipped_ingress_entries",
			"The number of ingress entries skipped by feed-dns, such as being outside of the Route53 hosted zone.")
	})
}


<|start_filename|>dns/dns_updater.go<|end_filename|>
package dns

import (
	"fmt"
	"strings"

	"github.com/aws/aws-sdk-go/service/route53"
	log "github.com/sirupsen/logrus"
	"github.com/sky-uk/feed/controller"
	"github.com/sky-uk/feed/dns/adapter"
	"github.com/sky-uk/feed/dns/r53"
)

type hostToIngress map[string]controller.IngressEntry

type updater struct {
	r53                 r53.Route53Client
	schemeToFrontendMap map[string]adapter.DNSDetails
	domain              string
	lbAdapter           adapter.FrontendAdapter
}

// New creates an updater for dns
func New(hostedZoneID string, lbAdapter adapter.FrontendAdapter, retries int) controller.Updater {
	initMetrics()

	return &updater{
		r53:                 r53.New(hostedZoneID, retries),
		lbAdapter:           lbAdapter,
		schemeToFrontendMap: make(map[string]adapter.DNSDetails),
	}
}

func (u *updater) String() string {
	return "route53 updater"
}

func (u *updater) Start() error {
	log.Info("Starting dns updater")

	schemeToFrontendMap, err := u.lbAdapter.Initialise()
	if err != nil {
		return err
	}
	u.schemeToFrontendMap = schemeToFrontendMap

	domain, err := u.r53.GetHostedZoneDomain()
	if err != nil {
		return fmt.Errorf("unable to get domain for hosted zone: %v", err)
	}
	u.domain = domain

	log.Info("Dns updater started")
	return nil
}

func (u *updater) Stop() error {
	return nil
}

func (u *updater) Health() error {
	return nil
}

func (u *updater) Update(entries controller.IngressEntries) error {
	route53Records, err := u.r53.GetRecords()
	if err != nil {
		log.Warn("Unable to get records from Route53. Not updating Route53.", err)
		failedCount.Inc()
		return err
	}

	// Flatten Alias (A) and CNAME records into a common structure
	records := u.consolidateRecordsFromRoute53(route53Records)

	records = u.determineManagedRecordSets(records)
	recordsGauge.Set(float64(len(records)))

	changes := u.calculateChanges(records, entries)

	updateCount.Add(float64(len(changes)))

	err = u.r53.UpdateRecordSets(changes)
	if err != nil {
		failedCount.Inc()
		return fmt.Errorf("unable to update record sets: %v", err)
	}

	return nil
}

func (u *updater) consolidateRecordsFromRoute53(rrs []*route53.ResourceRecordSet) []adapter.ConsolidatedRecord {
	var records []adapter.ConsolidatedRecord

	for _, recordSet := range rrs {
		if record, managed := u.lbAdapter.IsManaged(recordSet); managed {
			records = append(records, *record)
		}
	}

	return records
}

func (u *updater) determineManagedRecordSets(rrs []adapter.ConsolidatedRecord) []adapter.ConsolidatedRecord {
	managedLBs := make(map[string]bool)
	for _, dns := range u.schemeToFrontendMap {
		managedLBs[dns.DNSName] = true
	}
	var managed []adapter.ConsolidatedRecord
	var nonManaged []string
	for _, rec := range rrs {
		if rec.Name != "" && managedLBs[rec.PointsTo] {
			managed = append(managed, rec)
		} else {
			nonManaged = append(nonManaged, rec.Name)
		}
	}

	if len(nonManaged) > 0 {
		log.Infof("Filtered %d non-managed resource record sets: %v", len(nonManaged), nonManaged)
	}

	return managed
}

func (u *updater) calculateChanges(originalRecords []adapter.ConsolidatedRecord,
	entries controller.IngressEntries) []*route53.Change {

	log.Infof("Current %s records: %d", u.domain, len(originalRecords))
	log.Debugf("Current %s record set: %v", u.domain, originalRecords)
	log.Debug("Processing ingress update: ", entries)

	hostToIngress, skipped := u.indexByHost(entries)
	changes, skipped2 := u.createChanges(hostToIngress, originalRecords)

	skipped = append(skipped, skipped2...)

	if len(skipped) > 0 {
		log.Warnf("%d skipped entries for zone '%s': %v",
			len(skipped), u.domain, skipped)
	}

	log.Debug("Host to ingress entry: ", hostToIngress)
	log.Infof("Calculated changes to dns: %v", changes)
	return changes
}

func (u *updater) indexByHost(entries []controller.IngressEntry) (hostToIngress, []string) {
	var skipped []string
	mapping := make(hostToIngress)

	for _, entry := range entries {
		log.Debugf("Processing entry %v", entry)
		// Ingress entries in k8s aren't allowed to have the . on the end.
		// AWS adds it regardless of whether you specify it.
		hostNameWithPeriod := entry.Host + "."

		log.Debugf("Checking if ingress entry hostname %s is in domain %s", hostNameWithPeriod, u.domain)
		if !strings.HasSuffix(hostNameWithPeriod, "."+u.domain) {
			skipped = append(skipped, entry.NamespaceName()+":host:"+hostNameWithPeriod)
			skippedCount.Inc()
			continue
		}

		if previous, exists := mapping[hostNameWithPeriod]; exists {
			if previous.LbScheme != entry.LbScheme {
				skipped = append(skipped, entry.NamespaceName()+":conflicting-scheme:"+entry.LbScheme)
				skippedCount.Inc()
			}
		} else {
			mapping[hostNameWithPeriod] = entry
		}
	}

	return mapping, skipped
}

func (u *updater) createChanges(hostToIngress hostToIngress,
	originalRecords []adapter.ConsolidatedRecord) ([]*route53.Change, []string) {

	type recordKey struct{ host, elbDNSName string }
	var changes []*route53.Change
	indexedRecords := make(map[recordKey]adapter.ConsolidatedRecord)
	for _, rec := range originalRecords {
		indexedRecords[recordKey{rec.Name, rec.PointsTo}] = rec
	}

	var skipped []string
	for host, entry := range hostToIngress {
		dnsDetails, exists := u.schemeToFrontendMap[entry.LbScheme]
		if !exists {
			skipped = append(skipped, entry.NamespaceName()+":scheme:"+entry.LbScheme)
			skippedCount.Inc()
			continue
		}

		existingRecord, recordExists := indexedRecords[recordKey{host, dnsDetails.DNSName}]
		change := u.lbAdapter.CreateChange("UPSERT", host, dnsDetails, recordExists, &existingRecord)
		if change != nil {
			changes = append(changes, change)
		}
	}

	for _, rec := range originalRecords {
		if _, contains := hostToIngress[rec.Name]; !contains {
			changes = append(changes, u.lbAdapter.CreateChange("DELETE", rec.Name, adapter.DNSDetails{
				DNSName:      rec.PointsTo,
				HostedZoneID: rec.AliasHostedZone,
			}, false, nil))
		}
	}

	return changes, skipped
}


<|start_filename|>dns/adapter/frontend_adapter.go<|end_filename|>
package adapter

import "github.com/aws/aws-sdk-go/service/route53"

// FrontendAdapter defines operations which vary based on the type of load balancer being used for ingress.
type FrontendAdapter interface {
	Initialise() (map[string]DNSDetails, error)
	CreateChange(action string, host string, details DNSDetails, recordExists bool, existingRecord *ConsolidatedRecord) *route53.Change
	IsManaged(*route53.ResourceRecordSet) (*ConsolidatedRecord, bool)
}

// DNSDetails defines a DNS name and, optionally, how it maps to an AWS Route53 zone
type DNSDetails struct {
	DNSName      string
	HostedZoneID string
}

// ConsolidatedRecord describes how a DNS name maps to a static load balancer or AWS ELBs or ALBs.
type ConsolidatedRecord struct {
	Name            string
	PointsTo        string
	AliasHostedZone string
	TTL             int64
}
