Skip to content

Commit 7e73286

Browse files
joelanfordjoelanford
committed
include network policy for all configmap and grpc catalogsources
Signed-off-by: Joe Lanford <[email protected]>
1 parent c6ef935 commit 7e73286

File tree

12 files changed

+732
-95
lines changed

12 files changed

+732
-95
lines changed

‎pkg/controller/operators/catalog/operator.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"google.golang.org/grpc/connectivity"
1919
batchv1 "k8s.io/api/batch/v1"
2020
corev1 "k8s.io/api/core/v1"
21+
networkingv1 "k8s.io/api/networking/v1"
2122
rbacv1 "k8s.io/api/rbac/v1"
2223
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
2324
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -38,6 +39,7 @@ import (
3839
"k8s.io/apimachinery/pkg/util/yaml"
3940
batchv1applyconfigurations "k8s.io/client-go/applyconfigurations/batch/v1"
4041
corev1applyconfigurations "k8s.io/client-go/applyconfigurations/core/v1"
42+
networkingv1applyconfigurations "k8s.io/client-go/applyconfigurations/networking/v1"
4143
rbacv1applyconfigurations "k8s.io/client-go/applyconfigurations/rbac/v1"
4244
"k8s.io/client-go/dynamic"
4345
"k8s.io/client-go/informers"
@@ -600,6 +602,23 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
600602
}
601603
}
602604

605+
// Wire NetworkPolicies
606+
networkPolicyInformer := k8sInformerFactory.Networking().V1().NetworkPolicies()
607+
op.lister.NetworkingV1().RegisterNetworkPolicyLister(metav1.NamespaceAll, networkPolicyInformer.Lister())
608+
sharedIndexInformers = append(sharedIndexInformers, networkPolicyInformer.Informer())
609+
610+
networkPoliciesGVR := networkingv1.SchemeGroupVersion.WithResource("networkpolicies")
611+
if err := labelObjects(networkPoliciesGVR, networkPolicyInformer.Informer(), labeller.ObjectLabeler[*networkingv1.NetworkPolicy, *networkingv1applyconfigurations.NetworkPolicyApplyConfiguration](
612+
ctx, op.logger, labeller.Filter(networkPoliciesGVR),
613+
networkPolicyInformer.Lister().List,
614+
networkingv1applyconfigurations.NetworkPolicy,
615+
func(namespace string, ctx context.Context, cfg *networkingv1applyconfigurations.NetworkPolicyApplyConfiguration, opts metav1.ApplyOptions) (*networkingv1.NetworkPolicy, error) {
616+
return op.opClient.KubernetesInterface().NetworkingV1().NetworkPolicies(namespace).Apply(ctx, cfg, opts)
617+
},
618+
)); err != nil {
619+
return nil, err
620+
}
621+
603622
// Wire Pods for CatalogSource
604623
catsrcReq, err := labels.NewRequirement(reconciler.CatalogSourceLabelKey, selection.Exists, nil)
605624
if err != nil {

‎pkg/controller/registry/reconciler/configmap.go‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
pkgerrors "github.com/pkg/errors"
1212
"github.com/sirupsen/logrus"
1313
corev1 "k8s.io/api/core/v1"
14+
networkingv1 "k8s.io/api/networking/v1"
1415
rbacv1 "k8s.io/api/rbac/v1"
1516
apierrors "k8s.io/apimachinery/pkg/api/errors"
1617
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -121,6 +122,36 @@ func (s *configMapCatalogSourceDecorator) Pod(image string, defaultPodSecurityCo
121122
ownerutil.AddOwner(pod, s.CatalogSource, false, true)
122123
return pod, nil
123124
}
125+
func (s *configMapCatalogSourceDecorator) NetworkPolicy() *networkingv1.NetworkPolicy {
126+
np := &networkingv1.NetworkPolicy{
127+
ObjectMeta: metav1.ObjectMeta{
128+
Name: s.GetName(),
129+
Namespace: s.GetNamespace(),
130+
Labels: map[string]string{
131+
install.OLMManagedLabelKey: install.OLMManagedLabelValue,
132+
CatalogSourceLabelKey: s.GetName(),
133+
},
134+
},
135+
Spec: networkingv1.NetworkPolicySpec{
136+
PodSelector: metav1.LabelSelector{
137+
MatchLabels: s.Labels(),
138+
},
139+
PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress},
140+
Ingress: []networkingv1.NetworkPolicyIngressRule{
141+
{
142+
Ports: []networkingv1.NetworkPolicyPort{
143+
{
144+
Protocol: ptr.To(corev1.ProtocolTCP),
145+
Port: ptr.To(intstr.FromInt32(50051)),
146+
},
147+
},
148+
},
149+
},
150+
},
151+
}
152+
ownerutil.AddOwner(np, s.CatalogSource, false, false)
153+
return np
154+
}
124155

125156
func (s *configMapCatalogSourceDecorator) ServiceAccount() *corev1.ServiceAccount {
126157
sa := &corev1.ServiceAccount{
@@ -210,6 +241,16 @@ func (c *ConfigMapRegistryReconciler) currentService(source configMapCatalogSour
210241
return service, nil
211242
}
212243

244+
func (c *ConfigMapRegistryReconciler) currentNetworkPolicy(source configMapCatalogSourceDecorator) *networkingv1.NetworkPolicy {
245+
npName := source.NetworkPolicy().GetName()
246+
np, err := c.Lister.NetworkingV1().NetworkPolicyLister().NetworkPolicies(source.GetNamespace()).Get(npName)
247+
if err != nil {
248+
logrus.WithField("networkPolicy", npName).WithError(err).Debug("couldn't find network policy in cache")
249+
return nil
250+
}
251+
return np
252+
}
253+
213254
func (c *ConfigMapRegistryReconciler) currentServiceAccount(source configMapCatalogSourceDecorator) *corev1.ServiceAccount {
214255
serviceAccountName := source.ServiceAccount().GetName()
215256
serviceAccount, err := c.Lister.CoreV1().ServiceAccountLister().ServiceAccounts(source.GetNamespace()).Get(serviceAccountName)
@@ -328,6 +369,9 @@ func (c *ConfigMapRegistryReconciler) EnsureRegistryServer(logger *logrus.Entry,
328369
}
329370

330371
//TODO: if any of these error out, we should write a status back (possibly set RegistryServiceStatus to nil so they get recreated)
372+
if err := c.ensureNetworkPolicy(source); err != nil {
373+
return pkgerrors.Wrapf(err, "error ensuring network policy: %s", source.GetName())
374+
}
331375
if err := c.ensureServiceAccount(source, overwrite); err != nil {
332376
return pkgerrors.Wrapf(err, "error ensuring service account: %s", source.serviceAccountName())
333377
}
@@ -365,6 +409,20 @@ func (c *ConfigMapRegistryReconciler) EnsureRegistryServer(logger *logrus.Entry,
365409
return nil
366410
}
367411

412+
func (c *ConfigMapRegistryReconciler) ensureNetworkPolicy(source configMapCatalogSourceDecorator) error {
413+
networkPolicy := source.NetworkPolicy()
414+
if currentNetworkPolicy := c.currentNetworkPolicy(source); currentNetworkPolicy != nil {
415+
if sanitizedDeepEqual(networkPolicy, currentNetworkPolicy) {
416+
return nil
417+
}
418+
if err := c.OpClient.DeleteNetworkPolicy(networkPolicy.GetNamespace(), networkPolicy.GetName(), metav1.NewDeleteOptions(0)); err != nil && !apierrors.IsNotFound(err) {
419+
return err
420+
}
421+
}
422+
_, err := c.OpClient.CreateNetworkPolicy(networkPolicy)
423+
return err
424+
}
425+
368426
func (c *ConfigMapRegistryReconciler) ensureServiceAccount(source configMapCatalogSourceDecorator, overwrite bool) error {
369427
serviceAccount := source.ServiceAccount()
370428
if c.currentServiceAccount(source) != nil {
@@ -497,6 +555,18 @@ func (c *ConfigMapRegistryReconciler) CheckRegistryServer(logger *logrus.Entry,
497555
// Check on registry resources
498556
// TODO: more complex checks for resources
499557
// TODO: add gRPC health check
558+
np := c.currentNetworkPolicy(source)
559+
if np == nil {
560+
logger.Error("registry service not healthy: could not get network policy")
561+
healthy = false
562+
return
563+
}
564+
if !sanitizedDeepEqual(source.NetworkPolicy(), np) {
565+
logger.Error("registry service not healthy: unexpected network policy")
566+
healthy = false
567+
return
568+
}
569+
500570
service, err := c.currentService(source)
501571
if err != nil {
502572
return false, err

‎pkg/controller/registry/reconciler/configmap_test.go‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"github.com/sirupsen/logrus"
1313
"github.com/stretchr/testify/require"
1414
corev1 "k8s.io/api/core/v1"
15+
networkingv1 "k8s.io/api/networking/v1"
1516
rbacv1 "k8s.io/api/rbac/v1"
1617
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
1718
"k8s.io/apimachinery/pkg/api/meta"
@@ -83,6 +84,7 @@ func fakeReconcilerFactory(t *testing.T, stopc <-chan struct{}, options ...fakeR
8384
serviceInformer := informerFactory.Core().V1().Services()
8485
podInformer := informerFactory.Core().V1().Pods()
8586
configMapInformer := informerFactory.Core().V1().ConfigMaps()
87+
networkPolicyInformer := informerFactory.Networking().V1().NetworkPolicies()
8688

8789
registryInformers := []cache.SharedIndexInformer{
8890
roleInformer.Informer(),
@@ -91,6 +93,7 @@ func fakeReconcilerFactory(t *testing.T, stopc <-chan struct{}, options ...fakeR
9193
serviceInformer.Informer(),
9294
podInformer.Informer(),
9395
configMapInformer.Informer(),
96+
networkPolicyInformer.Informer(),
9497
}
9598

9699
lister := operatorlister.NewLister()
@@ -100,6 +103,7 @@ func fakeReconcilerFactory(t *testing.T, stopc <-chan struct{}, options ...fakeR
100103
lister.CoreV1().RegisterServiceLister(testNamespace, serviceInformer.Lister())
101104
lister.CoreV1().RegisterPodLister(testNamespace, podInformer.Lister())
102105
lister.CoreV1().RegisterConfigMapLister(testNamespace, configMapInformer.Lister())
106+
lister.NetworkingV1().RegisterNetworkPolicyLister(testNamespace, networkPolicyInformer.Lister())
103107

104108
rec := &registryReconcilerFactory{
105109
now: config.now,
@@ -195,6 +199,7 @@ func objectsForCatalogSource(t *testing.T, catsrc *v1alpha1.CatalogSource) []run
195199
switch catsrc.Spec.SourceType {
196200
case v1alpha1.SourceTypeInternal, v1alpha1.SourceTypeConfigmap:
197201
decorated := configMapCatalogSourceDecorator{catsrc, runAsUser}
202+
np := decorated.NetworkPolicy()
198203
service, err := decorated.Service()
199204
if err != nil {
200205
t.Fatal(err)
@@ -205,13 +210,15 @@ func objectsForCatalogSource(t *testing.T, catsrc *v1alpha1.CatalogSource) []run
205210
t.Fatal(err)
206211
}
207212
objs = append(objs,
213+
np,
208214
pod,
209215
service,
210216
serviceAccount,
211217
)
212218
case v1alpha1.SourceTypeGrpc:
213219
if catsrc.Spec.Image != "" {
214220
decorated := grpcCatalogSourceDecorator{CatalogSource: catsrc, createPodAsUser: runAsUser, opmImage: ""}
221+
np := decorated.NetworkPolicy()
215222
serviceAccount := decorated.ServiceAccount()
216223
service, err := decorated.Service()
217224
if err != nil {
@@ -222,6 +229,7 @@ func objectsForCatalogSource(t *testing.T, catsrc *v1alpha1.CatalogSource) []run
222229
t.Fatal(err)
223230
}
224231
objs = append(objs,
232+
np,
225233
pod,
226234
service,
227235
serviceAccount,
@@ -342,6 +350,24 @@ func TestConfigMapRegistryReconciler(t *testing.T) {
342350
},
343351
},
344352
},
353+
{
354+
testName: "ExistingRegistry/BadNetworkPolicy",
355+
in: in{
356+
cluster: cluster{
357+
k8sObjs: append(setLabel(objectsForCatalogSource(t, validCatalogSource), &networkingv1.NetworkPolicy{}, CatalogSourceLabelKey, "wrongValue"), validConfigMap),
358+
},
359+
catsrc: validCatalogSource,
360+
},
361+
out: out{
362+
status: &v1alpha1.RegistryServiceStatus{
363+
CreatedAt: now(),
364+
Protocol: "grpc",
365+
ServiceName: "cool-catalog",
366+
ServiceNamespace: testNamespace,
367+
Port: "50051",
368+
},
369+
},
370+
},
345371
{
346372
testName: "ExistingRegistry/BadServiceAccount",
347373
in: in{
@@ -504,6 +530,11 @@ func TestConfigMapRegistryReconciler(t *testing.T) {
504530
require.Equal(t, pod.GetLabels(), outPod.GetLabels())
505531
require.Equal(t, pod.Spec, outPod.Spec)
506532

533+
np := decorated.NetworkPolicy()
534+
outNp, err := client.KubernetesInterface().NetworkingV1().NetworkPolicies(np.GetNamespace()).Get(context.TODO(), np.GetName(), metav1.GetOptions{})
535+
require.NoError(t, err)
536+
require.Equal(t, np, outNp)
537+
507538
service, err := decorated.Service()
508539
require.NoError(t, err)
509540
outService, err := client.KubernetesInterface().CoreV1().Services(service.GetNamespace()).Get(context.TODO(), service.GetName(), metav1.GetOptions{})

‎pkg/controller/registry/reconciler/grpc.go‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,12 @@ import (
1919
pkgerrors "github.com/pkg/errors"
2020
"github.com/sirupsen/logrus"
2121
corev1 "k8s.io/api/core/v1"
22+
networkingv1 "k8s.io/api/networking/v1"
2223
apierrors "k8s.io/apimachinery/pkg/api/errors"
2324
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2425
"k8s.io/apimachinery/pkg/labels"
2526
"k8s.io/apimachinery/pkg/util/intstr"
27+
"k8s.io/utils/ptr"
2628
)
2729

2830
const (
@@ -102,6 +104,37 @@ func (s *grpcCatalogSourceDecorator) Service() (*corev1.Service, error) {
102104
return svc, nil
103105
}
104106

107+
func (s *grpcCatalogSourceDecorator) NetworkPolicy() *networkingv1.NetworkPolicy {
108+
np := &networkingv1.NetworkPolicy{
109+
ObjectMeta: metav1.ObjectMeta{
110+
Name: s.GetName(),
111+
Namespace: s.GetNamespace(),
112+
Labels: map[string]string{
113+
install.OLMManagedLabelKey: install.OLMManagedLabelValue,
114+
CatalogSourceLabelKey: s.GetName(),
115+
},
116+
},
117+
Spec: networkingv1.NetworkPolicySpec{
118+
PodSelector: metav1.LabelSelector{
119+
MatchLabels: s.Labels(),
120+
},
121+
PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress},
122+
Ingress: []networkingv1.NetworkPolicyIngressRule{
123+
{
124+
Ports: []networkingv1.NetworkPolicyPort{
125+
{
126+
Protocol: ptr.To(corev1.ProtocolTCP),
127+
Port: ptr.To(intstr.FromInt32(50051)),
128+
},
129+
},
130+
},
131+
},
132+
},
133+
}
134+
ownerutil.AddOwner(np, s.CatalogSource, false, false)
135+
return np
136+
}
137+
105138
func (s *grpcCatalogSourceDecorator) ServiceAccount() *corev1.ServiceAccount {
106139
var secrets []corev1.LocalObjectReference
107140
blockOwnerDeletion := true
@@ -153,6 +186,16 @@ type GrpcRegistryReconciler struct {
153186

154187
var _ RegistryReconciler = &GrpcRegistryReconciler{}
155188

189+
func (c *GrpcRegistryReconciler) currentNetworkPolicy(source grpcCatalogSourceDecorator) *networkingv1.NetworkPolicy {
190+
npName := source.NetworkPolicy().GetName()
191+
np, err := c.Lister.NetworkingV1().NetworkPolicyLister().NetworkPolicies(source.GetNamespace()).Get(npName)
192+
if err != nil {
193+
logrus.WithField("networkPolicy", npName).WithError(err).Debug("couldn't find network policy in cache")
194+
return nil
195+
}
196+
return np
197+
}
198+
156199
func (c *GrpcRegistryReconciler) currentService(source grpcCatalogSourceDecorator) (*corev1.Service, error) {
157200
protoService, err := source.Service()
158201
if err != nil {
@@ -261,6 +304,11 @@ func (c *GrpcRegistryReconciler) EnsureRegistryServer(logger *logrus.Entry, cata
261304
}
262305

263306
//TODO: if any of these error out, we should write a status back (possibly set RegistryServiceStatus to nil so they get recreated)
307+
if err := c.ensureNetworkPolicy(source); err != nil {
308+
logger.WithError(err).Error("error ensuring registry server: could not ensure registry network policy")
309+
return pkgerrors.Wrapf(err, "error ensuring network policy: %s", source.GetName())
310+
}
311+
264312
sa, err := c.ensureSA(source)
265313
if err != nil && !apierrors.IsAlreadyExists(err) {
266314
logger.WithError(err).Error("error ensuring registry server: could not ensure registry service account")
@@ -467,6 +515,20 @@ func (c *GrpcRegistryReconciler) ensureService(source grpcCatalogSourceDecorator
467515
return err
468516
}
469517

518+
func (c *GrpcRegistryReconciler) ensureNetworkPolicy(source grpcCatalogSourceDecorator) error {
519+
networkPolicy := source.NetworkPolicy()
520+
if currentNetworkPolicy := c.currentNetworkPolicy(source); currentNetworkPolicy != nil {
521+
if sanitizedDeepEqual(networkPolicy, currentNetworkPolicy) {
522+
return nil
523+
}
524+
if err := c.OpClient.DeleteNetworkPolicy(networkPolicy.GetNamespace(), networkPolicy.GetName(), metav1.NewDeleteOptions(0)); err != nil && !apierrors.IsNotFound(err) {
525+
return err
526+
}
527+
}
528+
_, err := c.OpClient.CreateNetworkPolicy(networkPolicy)
529+
return err
530+
}
531+
470532
func (c *GrpcRegistryReconciler) ensureSA(source grpcCatalogSourceDecorator) (*corev1.ServiceAccount, error) {
471533
sa := source.ServiceAccount()
472534
if _, err := c.OpClient.CreateServiceAccount(sa); err != nil {
@@ -606,6 +668,17 @@ func (c *GrpcRegistryReconciler) CheckRegistryServer(logger *logrus.Entry, catal
606668

607669
// Check on registry resources
608670
// TODO: add gRPC health check
671+
currentNetworkPolicy := c.currentNetworkPolicy(source)
672+
if currentNetworkPolicy == nil {
673+
logger.Error("registry service not healthy: could not get network policy")
674+
return false, nil
675+
}
676+
expectedNetworkPolicy := source.NetworkPolicy()
677+
if !sanitizedDeepEqual(expectedNetworkPolicy, currentNetworkPolicy) {
678+
logger.Error("registry service not healthy: unexpected network policy")
679+
return false, nil
680+
}
681+
609682
service, err := c.currentService(source)
610683
if err != nil {
611684
logger.WithError(err).Error("registry service not healthy: could not get current service")

0 commit comments

Comments
 (0)