Network Policy Agent - Initial commit #1
Conversation
| yum install -y coreutils kernel-devel elfutils-libelf-devel zlib-devel bpftool libbpf-devel && \ | ||
| yum clean all | ||
|
|
||
| COPY Makefile ./ |
There was a problem hiding this comment.
COPY Makefile ./ seem redundant with the COPY . ./
| COPY . ./ | ||
| RUN make build-bpf | ||
|
|
||
| # Use distroless as minimal base image to package the manager binary |
| # Image URL to use all building/pushing image targets | ||
| IMAGE ?= amazon/aws-network-policy-agent | ||
| VERSION ?= $(shell git describe --tags --always --dirty || echo "unknown") | ||
| IMAGE_NAME = $(IMAGE)$(IMAGE_ARCH_SUFFIX):$(VERSION) |
There was a problem hiding this comment.
where is the IMAGE_ARCH_SUFFIX defined?
| @@ -0,0 +1,139 @@ | |||
| /* | |||
There was a problem hiding this comment.
seems redundant definition with the network policy controller, i assume we'll switch to depends on the controller's crd once it's published.
|
|
||
| const ( | ||
| envLocalConntrackCacheCleanupPeriod = "CONNTRACK_CACHE_CLEANUP_PERIOD" | ||
| defaultLocalConntrackCacheCleanupPeriod = 300 |
There was a problem hiding this comment.
nit: better to name the variable xxxUnit e.g. defaultLocalConntrackCacheCleanupPeriodSeconds if it's numeric value without unit.
|
|
||
| // PolicyEndpointsReconciler reconciles a PolicyEndpoints object | ||
| type PolicyEndpointsReconciler struct { | ||
| K8sClient client.Client |
There was a problem hiding this comment.
nit:
- the usage of Public & Private members seems arbitrary, members like K8sClient /Scheme don't need to be public
- the class PolicyEndpointsReconciler itself don't need to be a public as well. (only NewPolicyEndpointsReconciler need to be public)
|
|
||
| r.nodeIP, err = imds.GetNodeAddress(enableIPv6) | ||
| if err != nil { | ||
| log.Error(err, "Unable to derive Node IP. Abort....") |
There was a problem hiding this comment.
we should return error here and skip the rest processing when error encountered
otherwise, this err might be overwrite to be nil and became ignored by the r.ebpfClient, err = ebpf.NewBpfClient
| } | ||
|
|
||
| conntrackTTL := r.getLocalConntrackCacheCleanupPeriod() | ||
| r.ebpfClient, err = ebpf.NewBpfClient(&r.policyEndpointeBPFContext, &r.IngressProgPodMap, |
There was a problem hiding this comment.
nit: this component's functionality apparently goes beyond a simple "bpf's client". It seems more like a eBPFNetworkEnforcer /EBPFNetworkPolicyManager/networkPolicyEbpfController (i'm bad at naming as well)
| func (r *PolicyEndpointsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { | ||
| r.Log.Info("Received a new reconcile request", "req", req) | ||
| if err := r.reconcile(ctx, req); err != nil { | ||
| r.Log.Info("Reconcile error - requeue the request", "err", err) |
There was a problem hiding this comment.
nit: better to be r.Log.Error, this comment applies to other similar error logs
| func (r *PolicyEndpointsReconciler) reconcile(ctx context.Context, req ctrl.Request) error { | ||
| policyEndpoint := &policyk8sawsv1.PolicyEndpoint{} | ||
| if err := r.K8sClient.Get(ctx, req.NamespacedName, policyEndpoint); err != nil { | ||
| if err = client.IgnoreNotFound(err); err == nil { |
There was a problem hiding this comment.
nit: better to use apierror.IsNotFoundError instead of client.IgnoreNotFound(err) when check whether error is not found. client.IgnoreNotFound(err) should be used with return statements only.
| return r.cleanUpPolicyEndpoint(ctx, req) | ||
| } | ||
| r.Log.Info("Unable to get policy endpoint spec", "policyendpoint", req.NamespacedName, "error", err) | ||
| return client.IgnoreNotFound(err) |
There was a problem hiding this comment.
the IgnoreNotFound calls seems unnecessary given it's already tested against not found.
| } | ||
|
|
||
| func (r *PolicyEndpointsReconciler) cleanUpPolicyEndpoint(ctx context.Context, req ctrl.Request) error { | ||
| r.Log.Info("Clean Up PolicyEndpoint resources for", "name:", req.NamespacedName.Name) |
There was a problem hiding this comment.
logs here seems need to log both name and namespace
| req.NamespacedName.Namespace) | ||
|
|
||
| start := time.Now() | ||
| duration := msSince(start) |
There was a problem hiding this comment.
seems this line is in wrong place, the duration here will always close to zero?
| duration := msSince(start) | ||
|
|
||
| if targetPods, ok := r.policyEndpointSelectorMap.Load(policyEndpointIdentifier); ok { | ||
| err := r.updatePods(ctx, req, targetPods.([]types.NamespacedName)) |
There was a problem hiding this comment.
nit: better to use
if err := ...; err != nil {
....
}
| return nil | ||
| } | ||
|
|
||
| func (r *PolicyEndpointsReconciler) updatePods(ctx context.Context, req ctrl.Request, |
There was a problem hiding this comment.
let's have some comments for this function.
nit: pick a more meaningful name than updatePods if possible. (e.g. updateEbpfProbesForPods)
| // 3. If there are no more active policies against this pod. Detach and delete the probes and | ||
| // corresponding BPF maps. We will also clean up eBPF pin paths under BPF FS. | ||
| for _, targetPod := range targetPods { | ||
| r.Log.Info("Updating Pod: ", "Name: ", targetPod.Name, "Namespace: ", targetPod.Namespace) |
There was a problem hiding this comment.
since we are using structured logging, we don't need the : in field names.
this comment applies to other logging statements as well.
| for _, targetPod := range targetPods { | ||
| r.Log.Info("Updating Pod: ", "Name: ", targetPod.Name, "Namespace: ", targetPod.Namespace) | ||
| podIdentifier := utils.GetPodIdentifier(targetPod.Name, targetPod.Namespace) | ||
| err = r.cleanupeBPFProbes(ctx, targetPod, podIdentifier, req.NamespacedName.Name) |
There was a problem hiding this comment.
the input to this function seems weird to me
- it accepts the same namespacedName of pod twice(targetPod/podIdentifier) in different format. why not put the podIdentifier calculation inside cleanupeBPFProbes?
- why only name of policyendpoint is passed in? (the namespace is not passed)
(disclaimer: i haven't read following code yet)
| err = r.cleanupeBPFProbes(ctx, targetPod, podIdentifier, req.NamespacedName.Name) | ||
| if err != nil { | ||
| r.Log.Info("Cleanup/Update unsuccessful for Pod ", "Name: ", targetPod.Name, "Namespace: ", targetPod.Namespace) | ||
| // we don't want to return an error right away but instead attempt to clean up all the pods |
There was a problem hiding this comment.
maybe use errors.Aggregate to aggregate errors?
https://github.com/kubernetes/apimachinery/blob/master/pkg/util/errors/errors.go
| // in the list before returning | ||
| } | ||
| } | ||
| return err |
There was a problem hiding this comment.
this don't work. imaging there is two pod, cleanup for first pod failed, and cleanup for second pod succeed, but we are returning nil instead of error here.
consider use errors.Aggregate here:
https://github.com/kubernetes/apimachinery/blob/master/pkg/util/errors/errors.go
| r.Log.Info("Processing: ", "Policy Endpoint Identifer: ", policyEndpointIdentifier) | ||
|
|
||
| start := time.Now() | ||
| duration := msSince(start) |
| r.Log.Info("Error updating the older pods") //TODO - we need to continue and not bail out.retry? | ||
| } | ||
|
|
||
| for podIdentifier, _ := range podIdentifiers { |
There was a problem hiding this comment.
seem the bool value of podIdentifiers map is not used? if so, why return a map in the first place..
| for podIdentifier, _ := range podIdentifiers { | ||
| // Derive Ingress IPs from the PolicyEndpoint | ||
| ingressRules, egressRules, isIngressIsolated, isEgressIsolated, err := r.deriveIngressAndEgressFirewallRules(ctx, policyEndpoint, podIdentifier, | ||
| policyEndpoint.ObjectMeta.Namespace) |
There was a problem hiding this comment.
nit: you can use policyEndpoint.Namespace directly
|
|
||
| for podIdentifier, _ := range podIdentifiers { | ||
| // Derive Ingress IPs from the PolicyEndpoint | ||
| ingressRules, egressRules, isIngressIsolated, isEgressIsolated, err := r.deriveIngressAndEgressFirewallRules(ctx, policyEndpoint, podIdentifier, |
There was a problem hiding this comment.
similarly, this function interface is werid, you passed both policyEndpoint, policyEndpoint.Namespace to it. unless there are cases where a different namespace than policyEndpoint.Namespace will be passed, it's better to get namespace from policyEndpoint within deriveIngressAndEgressFirewallRules
| return nil | ||
| } | ||
|
|
||
| func (r *PolicyEndpointsReconciler) configureeBPFProbes(ctx context.Context, podIdentifier string, |
There was a problem hiding this comment.
the mixed usage of podIdentifier and types.NamespacedName really hurts readability a lot...
can we stick to one of them in all controller code to present pod/policyendpoint's id? e.g. you should be able to use types.NamespacedName as key to sync.map.
Issue #, if available:
Initial commit
Description of changes:
Initial commit for AWS Network Policy agent
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.