opaque_ports_watcher_test.go

Documentation: github.com/linkerd/linkerd2/controller/api/destination/watcher

     1  package watcher
     2  
     3  import (
     4  	"testing"
     5  
     6  	"github.com/linkerd/linkerd2/controller/k8s"
     7  	logging "github.com/sirupsen/logrus"
     8  	corev1 "k8s.io/api/core/v1"
     9  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    10  )
    11  
    12  var (
    13  	testNS = `
    14  apiVersion: v1
    15  kind: Namespace
    16  metadata:
    17    name: ns`
    18  	testNSObject = corev1.Namespace{
    19  		ObjectMeta: metav1.ObjectMeta{
    20  			Name: "ns",
    21  		},
    22  	}
    23  	baseService = `
    24  apiVersion: v1
    25  kind: Service
    26  metadata:
    27    name: svc
    28    namespace: ns`
    29  	baseServiceObject = corev1.Service{
    30  		ObjectMeta: metav1.ObjectMeta{
    31  			Name:      "svc",
    32  			Namespace: "ns",
    33  		},
    34  		Spec: corev1.ServiceSpec{
    35  			Ports: []corev1.ServicePort{{Port: 8080}},
    36  		},
    37  	}
    38  	opaqueService = `
    39  apiVersion: v1
    40  kind: Service
    41  metadata:
    42    name: svc
    43    namespace: ns
    44    annotations:
    45      config.linkerd.io/opaque-ports: "3306"`
    46  	opaqueServiceObject = corev1.Service{
    47  		ObjectMeta: metav1.ObjectMeta{
    48  			Name:        "svc",
    49  			Namespace:   "ns",
    50  			Annotations: map[string]string{"config.linkerd.io/opaque-ports": "3306"},
    51  		},
    52  		Spec: corev1.ServiceSpec{
    53  			Ports: []corev1.ServicePort{{Port: 3306}},
    54  		},
    55  	}
    56  	opaqueServiceMultiPort = `
    57  apiVersion: v1
    58  kind: Service
    59  metadata:
    60    name: svc
    61    namespace: ns
    62    annotations:
    63      config.linkerd.io/opaque-ports: "3306, 665"`
    64  	opaqueServiceMultiPortObject = corev1.Service{
    65  		ObjectMeta: metav1.ObjectMeta{
    66  			Name:        "svc",
    67  			Namespace:   "ns",
    68  			Annotations: map[string]string{"config.linkerd.io/opaque-ports": "3306, 665"},
    69  		},
    70  		Spec: corev1.ServiceSpec{
    71  			Ports: []corev1.ServicePort{{Port: 3306}, {Port: 665}},
    72  		},
    73  	}
    74  	explicitlyNotOpaqueService = `
    75  apiVersion: v1
    76  kind: Service
    77  metadata:
    78    name: svc
    79    namespace: ns
    80    annotations:
    81      config.linkerd.io/opaque-ports: ""`
    82  	explicitlyNotOpaqueServiceObject = corev1.Service{
    83  		ObjectMeta: metav1.ObjectMeta{
    84  			Name:        "svc",
    85  			Namespace:   "ns",
    86  			Annotations: map[string]string{"config.linkerd.io/opaque-ports": ""},
    87  		},
    88  		Spec: corev1.ServiceSpec{
    89  			Ports: []corev1.ServicePort{{Port: 3306}},
    90  		},
    91  	}
    92  )
    93  
    94  type testOpaquePortsListener struct {
    95  	updates []map[uint32]struct{}
    96  }
    97  
    98  func newTestOpaquePortsListener() *testOpaquePortsListener {
    99  	return &testOpaquePortsListener{
   100  		updates: []map[uint32]struct{}{},
   101  	}
   102  }
   103  
   104  func (bopl *testOpaquePortsListener) UpdateService(ports map[uint32]struct{}) {
   105  	bopl.updates = append(bopl.updates, ports)
   106  }
   107  
   108  func TestOpaquePortsWatcher(t *testing.T) {
   109  	defaultOpaquePorts := map[uint32]struct{}{
   110  		25:    {},
   111  		443:   {},
   112  		587:   {},
   113  		3306:  {},
   114  		5432:  {},
   115  		11211: {},
   116  	}
   117  
   118  	for _, tt := range []struct {
   119  		name                string
   120  		initialState        []string
   121  		nsObject            interface{}
   122  		svcObject           interface{}
   123  		service             ServiceID
   124  		expectedOpaquePorts []map[uint32]struct{}
   125  	}{
   126  		{
   127  			name:         "namespace and service",
   128  			initialState: []string{testNS, baseService},
   129  			nsObject:     &testNSObject,
   130  			svcObject:    &baseServiceObject,
   131  			service: ServiceID{
   132  				Name:      "svc",
   133  				Namespace: "ns",
   134  			},
   135  			// 1. default opaque ports
   136  			// 2. svc updated: no update
   137  			// 3. svc deleted: no update
   138  			// 4. svc created: ?
   139  			expectedOpaquePorts: []map[uint32]struct{}{{11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}},
   140  		},
   141  		{
   142  			name:         "namespace with opaque service",
   143  			initialState: []string{testNS, opaqueService},
   144  			nsObject:     &testNSObject,
   145  			svcObject:    &opaqueServiceObject,
   146  			service: ServiceID{
   147  				Name:      "svc",
   148  				Namespace: "ns",
   149  			},
   150  			// 1: svc annotation 3306
   151  			// 2: svc updated: no update
   152  			// 2: svc deleted: update with default ports
   153  			// 3. svc created: update with port 3306
   154  			expectedOpaquePorts: []map[uint32]struct{}{{3306: {}}, {11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}, {3306: {}}},
   155  		},
   156  		{
   157  			name:         "namespace with multi port opaque service",
   158  			initialState: []string{testNS, opaqueServiceMultiPort},
   159  			nsObject:     &testNSObject,
   160  			svcObject:    &opaqueServiceMultiPortObject,
   161  			service: ServiceID{
   162  				Name:      "svc",
   163  				Namespace: "ns",
   164  			},
   165  			// 1: svc annotation 3306, 665 (with whitespace)
   166  			// 2: svc updated: no update
   167  			// 2: svc deleted: update with default ports
   168  			// 3. svc created: update with port 3306, 665
   169  			expectedOpaquePorts: []map[uint32]struct{}{{3306: {}, 665: {}}, {11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}, {3306: {}, 665: {}}},
   170  		},
   171  		{
   172  			name:         "namespace and service, create opaque service",
   173  			initialState: []string{testNS, baseService},
   174  			nsObject:     &testNSObject,
   175  			svcObject:    &opaqueServiceObject,
   176  			service: ServiceID{
   177  				Name:      "svc",
   178  				Namespace: "ns",
   179  			},
   180  			// 1: default opaque ports
   181  			// 2: svc updated: update with port 3306
   182  			// 3: svc deleted: update with default ports
   183  			// 4. svc created: update with port 3306
   184  			expectedOpaquePorts: []map[uint32]struct{}{{11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}, {3306: {}}, {11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}, {3306: {}}},
   185  		},
   186  		{
   187  			name:         "namespace and opaque service, create base service",
   188  			initialState: []string{testNS, opaqueService},
   189  			nsObject:     &testNSObject,
   190  			svcObject:    &baseServiceObject,
   191  			service: ServiceID{
   192  				Name:      "svc",
   193  				Namespace: "ns",
   194  			},
   195  			// 1: svc annotation 3306
   196  			// 2. svc updated: update with default ports
   197  			// 3. svc deleted: no update
   198  			// 4. svc added: no update
   199  			expectedOpaquePorts: []map[uint32]struct{}{{3306: {}}, {11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}},
   200  		},
   201  		{
   202  			name:         "namespace and explicitly not opaque service, create explicitly not opaque service",
   203  			initialState: []string{testNS, explicitlyNotOpaqueService},
   204  			nsObject:     &testNSObject,
   205  			svcObject:    &explicitlyNotOpaqueServiceObject,
   206  			service: ServiceID{
   207  				Name:      "svc",
   208  				Namespace: "ns",
   209  			},
   210  			// 1: svc annotation empty
   211  			// 2. svc updated: no update
   212  			// 3. svc deleted: update with default ports
   213  			// 4. svc added: update with no ports
   214  			expectedOpaquePorts: []map[uint32]struct{}{{}, {11211: {}, 25: {}, 3306: {}, 443: {}, 5432: {}, 587: {}}, {}},
   215  		},
   216  	} {
   217  		k8sAPI, err := k8s.NewFakeAPI(tt.initialState...)
   218  		if err != nil {
   219  			t.Fatalf("NewFakeAPI returned an error: %s", err)
   220  		}
   221  		watcher, err := NewOpaquePortsWatcher(k8sAPI, logging.WithField("test", t.Name()), defaultOpaquePorts)
   222  		if err != nil {
   223  			t.Fatalf("can't create opaque ports watcher: %s", err)
   224  		}
   225  		k8sAPI.Sync(nil)
   226  		listener := newTestOpaquePortsListener()
   227  		watcher.Subscribe(tt.service, listener)
   228  		watcher.addService(tt.svcObject)
   229  		watcher.deleteService(tt.svcObject)
   230  		watcher.addService(tt.svcObject)
   231  		testCompare(t, tt.expectedOpaquePorts, listener.updates)
   232  	}
   233  }
   234
View as plain text