analyze: start parsing anps and banp from kube server or path

cmd/policy-assistant/pkg/cli/analyze.go

@@ -4,8 +4,12 @@ import (
 	"fmt"
 	"github.com/mattfenwick/cyclonus/examples"
 	"github.com/mattfenwick/cyclonus/pkg/kube/netpol"
+	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/network-policy-api/apis/v1alpha1"
 	"strings"
+	"time"
 
 	"github.com/mattfenwick/collections/pkg/json"
 	"github.com/mattfenwick/cyclonus/pkg/connectivity/probe"
@@ -38,6 +42,8 @@ var AllModes = []string{
 	ProbeMode,
 }
 
+const DefaultTimeout = 3 * time.Minute
+
 type AnalyzeArgs struct {
 	AllNamespaces      bool
 	Namespaces         []string
@@ -56,6 +62,8 @@ type AnalyzeArgs struct {
 
 	// synthetic probe
 	ProbePath string
+
+	Timeout time.Duration
 }
 
 func SetupAnalyzeCommand() *cobra.Command {
@@ -82,6 +90,7 @@ func SetupAnalyzeCommand() *cobra.Command {
 	command.Flags().StringVar(&args.TargetPodPath, "target-pod-path", "", "path to json target pod file -- json array of dicts")
 	command.Flags().StringVar(&args.TrafficPath, "traffic-path", "", "path to json traffic file, containing of a list of traffic objects")
 	command.Flags().StringVar(&args.ProbePath, "probe-path", "", "path to json model file for synthetic probe")
+	command.Flags().DurationVar(&args.Timeout, "kube-client-timeout", DefaultTimeout, "kube client timeout")
 
 	return command
 }
@@ -90,9 +99,10 @@ func RunAnalyzeCommand(args *AnalyzeArgs) {
 	// 1. read policies from kube
 	var kubePolicies []*networkingv1.NetworkPolicy
 	var kubeANPs []*v1alpha1.AdminNetworkPolicy
-	var kubeBANPs *v1alpha1.BaselineAdminNetworkPolicy
+	var kubeBANP *v1alpha1.BaselineAdminNetworkPolicy
 	var kubePods []v1.Pod
 	var kubeNamespaces []v1.Namespace
+	var netpolErr, anpErr, banpErr error
 	if args.AllNamespaces || len(args.Namespaces) > 0 {
 		kubeClient, err := kube.NewKubernetesForContext(args.Context)
 		utils.DoOrDie(err)
@@ -104,31 +114,48 @@ func RunAnalyzeCommand(args *AnalyzeArgs) {
 			kubeNamespaces = nsList.Items
 			namespaces = []string{v1.NamespaceAll}
 		}
-		kubePolicies, err = kube.ReadNetworkPoliciesFromKube(kubeClient, namespaces)
-		if err != nil {
+
+		includeANPS, includeBANPSs := shouldIncludeANPandBANP(kubeClient.ClientSet)
+
+		ctx, cancel := context.WithTimeout(context.TODO(), args.Timeout)
+		defer cancel()
+
+		kubePolicies, kubeANPs, kubeBANP, netpolErr, anpErr, banpErr = kube.ReadNetworkPoliciesFromKube(ctx, kubeClient, namespaces, includeANPS, includeBANPSs)
+
+		if netpolErr != nil {
 			logrus.Errorf("unable to read network policies from kube, ns '%s': %+v", namespaces, err)
 		}
-		kubePods, err = kube.GetPodsInNamespaces(kubeClient, namespaces)
-		if err != nil {
-			logrus.Errorf("unable to read pods from kube, ns '%s': %+v", namespaces, err)
+		if anpErr != nil {
+			logrus.Errorf("Unable to fetch admin network policies: %s \n", anpErr)
+		}
+		if banpErr != nil {
+			logrus.Errorf("Unable to fetch base admin network policies: %s \n", banpErr)
 		}
 	}
 	// 2. read policies from file
 	if args.PolicyPath != "" {
-		policiesFromPath, err := kube.ReadNetworkPoliciesFromPath(args.PolicyPath)
+		policiesFromPath, anpsFromPath, banpFromPath, err := kube.ReadNetworkPoliciesFromPath(args.PolicyPath)
 		utils.DoOrDie(err)
 		kubePolicies = append(kubePolicies, policiesFromPath...)
+		kubeANPs = append(kubeANPs, anpsFromPath...)
+		if banpFromPath != nil && kubeBANP != nil {
+			logrus.Debugf("More that one banp parsed - setting banp from file")
+		}
+		kubeBANP = banpFromPath
 	}
 	// 3. read example policies
 	if args.UseExamplePolicies {
 		kubePolicies = append(kubePolicies, netpol.AllExamples...)
 
-		kubeANPs = examples.CoreGressRulesCombinedANB
-		kubeBANPs = examples.CoreGressRulesCombinedBANB
+		kubeANPs = append(kubeANPs, examples.CoreGressRulesCombinedANB...)
+		if kubeBANP != nil {
+			logrus.Debugf("More that onew banp parsed - setting banp from the examples")
+		}
+		kubeBANP = examples.CoreGressRulesCombinedBANB
 	}
 
 	logrus.Debugf("parsed policies:\n%s", json.MustMarshalToString(kubePolicies))
-	policies := matcher.BuildV1AndV2NetPols(args.SimplifyPolicies, kubePolicies, kubeANPs, kubeBANPs)
+	policies := matcher.BuildV1AndV2NetPols(args.SimplifyPolicies, kubePolicies, kubeANPs, kubeBANP)
 
 	for _, mode := range args.Modes {
 		switch mode {
@@ -303,3 +330,28 @@ func ProbeSyntheticConnectivity(explainedPolicies *matcher.Policy, modelPath str
 	fmt.Printf("Egress:\n%s\n", simulatedProbe.RenderEgress())
 	fmt.Printf("Combined:\n%s\n\n\n", simulatedProbe.RenderTable())
 }
+
+func shouldIncludeANPandBANP(client *kubernetes.Clientset) (bool, bool) {
+	var includeANP, includeBANP bool
+	_, resources, _, err := client.DiscoveryClient.GroupsAndMaybeResources()
+	if err != nil {
+		logrus.Errorf("Unable to fetch all registered resources: %s", err)
+		return includeANP, includeBANP
+	}
+	gv := schema.GroupVersion{Group: "policy.networking.k8s.io", Version: "v1alpha1"}
+
+	if groupResources, ok := resources[gv]; ok {
+		for _, res := range groupResources.APIResources {
+			switch res.Kind {
+			case "AdminNetworkPolicy":
+				includeANP = true
+			case "BaselineAdminNetworkPolicy":
+				includeBANP = true
+			default:
+				continue
+			}
+		}
+	}
+
+	return includeANP, includeBANP
+}

cmd/policy-assistant/pkg/connectivity/testcasestate.go

@@ -1,6 +1,7 @@
 package connectivity
 
 import (
+	"context"
 	"time"
 
 	"github.com/mattfenwick/cyclonus/pkg/connectivity/probe"
@@ -140,7 +141,7 @@ func (t *TestCaseState) DeletePod(ns string, pod string) error {
 }
 
 func (t *TestCaseState) ReadPolicies(namespaces []string) error {
-	policies, err := kube.GetNetworkPoliciesInNamespaces(t.Kubernetes, namespaces)
+	policies, err := kube.GetNetworkPoliciesInNamespaces(context.TODO(), t.Kubernetes, namespaces)
 	if err != nil {
 		return err
 	}
@@ -322,7 +323,7 @@ func (t *TestCaseState) VerifyClusterState() error {
 		return err
 	}
 
-	policies, err := kube.GetNetworkPoliciesInNamespaces(t.Kubernetes, t.Resources.NamespacesSlice())
+	policies, err := kube.GetNetworkPoliciesInNamespaces(context.TODO(), t.Kubernetes, t.Resources.NamespacesSlice())
 	if err != nil {
 		return err
 	}

cmd/policy-assistant/pkg/kube/ikubernetes.go

@@ -1,13 +1,15 @@
 package kube
 
 import (
+	"context"
 	"fmt"
 	"github.com/mattfenwick/cyclonus/pkg/utils"
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"math/rand"
+	"sigs.k8s.io/network-policy-api/apis/v1alpha1"
 )
 
 type IKubernetes interface {
@@ -18,7 +20,7 @@ type IKubernetes interface {
 	GetAllNamespaces() (*v1.NamespaceList, error)
 
 	CreateNetworkPolicy(kubePolicy *networkingv1.NetworkPolicy) (*networkingv1.NetworkPolicy, error)
-	GetNetworkPoliciesInNamespace(namespace string) ([]networkingv1.NetworkPolicy, error)
+	GetNetworkPoliciesInNamespace(ctx context.Context, namespace string) ([]networkingv1.NetworkPolicy, error)
 	UpdateNetworkPolicy(kubePolicy *networkingv1.NetworkPolicy) (*networkingv1.NetworkPolicy, error)
 	DeleteNetworkPolicy(namespace string, name string) error
 	DeleteAllNetworkPoliciesInNamespace(namespace string) error
@@ -28,6 +30,16 @@ type IKubernetes interface {
 	DeleteService(namespace string, name string) error
 	GetServicesInNamespace(namespace string) ([]v1.Service, error)
 
+	GetAdminNetworkPolicies(ctx context.Context) ([]v1alpha1.AdminNetworkPolicy, error)
+	CreateAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.AdminNetworkPolicy) (*v1alpha1.AdminNetworkPolicy, error)
+	UpdateAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.AdminNetworkPolicy) (*v1alpha1.AdminNetworkPolicy, error)
+	DeleteAdminNetworkPolicy(ctx context.Context, name string) error
+
+	GetBaselineAdminNetworkPolicy(ctx context.Context) (*v1alpha1.BaselineAdminNetworkPolicy, error)
+	CreateBaselineAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.BaselineAdminNetworkPolicy) (*v1alpha1.BaselineAdminNetworkPolicy, error)
+	UpdateBaselineAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.BaselineAdminNetworkPolicy) (*v1alpha1.BaselineAdminNetworkPolicy, error)
+	DeleteBaselineAdminNetworkPolicy(ctx context.Context, name string) error
+
 	CreatePod(kubePod *v1.Pod) (*v1.Pod, error)
 	GetPod(namespace string, pod string) (*v1.Pod, error)
 	DeletePod(namespace string, pod string) error
@@ -37,10 +49,10 @@ type IKubernetes interface {
 	ExecuteRemoteCommand(namespace string, pod string, container string, command []string) (string, string, error, error)
 }
 
-func GetNetworkPoliciesInNamespaces(kubernetes IKubernetes, namespaces []string) ([]networkingv1.NetworkPolicy, error) {
+func GetNetworkPoliciesInNamespaces(ctx context.Context, kubernetes IKubernetes, namespaces []string) ([]networkingv1.NetworkPolicy, error) {
 	var allNetpols []networkingv1.NetworkPolicy
 	for _, ns := range namespaces {
-		netpols, err := kubernetes.GetNetworkPoliciesInNamespace(ns)
+		netpols, err := kubernetes.GetNetworkPoliciesInNamespace(ctx, ns)
 		if err != nil {
 			return nil, err
 		}
@@ -83,6 +95,14 @@ func GetServicesInNamespaces(kubernetes IKubernetes, namespaces []string) ([]v1.
 	return allServices, nil
 }
 
+func GetAdminNetworkPolicies(ctx context.Context, kubernetes IKubernetes) ([]v1alpha1.AdminNetworkPolicy, error) {
+	return kubernetes.GetAdminNetworkPolicies(ctx)
+}
+
+func GetBaselineAdminNetworkPolicy(ctx context.Context, kubernetes IKubernetes) (*v1alpha1.BaselineAdminNetworkPolicy, error) {
+	return kubernetes.GetBaselineAdminNetworkPolicy(ctx)
+}
+
 type MockNamespace struct {
 	NamespaceObject *v1.Namespace
 	Netpols         map[string]*networkingv1.NetworkPolicy
@@ -91,9 +111,14 @@ type MockNamespace struct {
 }
 
 type MockKubernetes struct {
-	Namespaces map[string]*MockNamespace
-	passRate   float64
-	podID      int
+	AdminNetworkPolicies        []v1alpha1.AdminNetworkPolicy
+	AdminNetworkPolicyError     error
+	BaselineNetworkPolicy       *v1alpha1.BaselineAdminNetworkPolicy
+	BaseAdminNetworkPolicyError error
+	Namespaces                  map[string]*MockNamespace
+	NetworkPolicyError          error
+	passRate                    float64
+	podID                       int
 }
 
 func NewMockKubernetes(passRate float64) *MockKubernetes {
@@ -191,7 +216,11 @@ func (m *MockKubernetes) DeleteNetworkPolicy(ns string, name string) error {
 	return nil
 }
 
-func (m *MockKubernetes) GetNetworkPoliciesInNamespace(namespace string) ([]networkingv1.NetworkPolicy, error) {
+func (m *MockKubernetes) GetNetworkPoliciesInNamespace(ctx context.Context, namespace string) ([]networkingv1.NetworkPolicy, error) {
+	if m.NetworkPolicyError != nil {
+		return nil, m.NetworkPolicyError
+	}
+
 	nsObject, err := m.getNamespaceObject(namespace)
 	if err != nil {
 		return nil, err
@@ -363,3 +392,37 @@ func (m *MockKubernetes) ExecuteRemoteCommand(namespace string, pod string, cont
 	}
 	return "", "", nil, nil
 }
+
+func (m *MockKubernetes) GetAdminNetworkPolicies(ctx context.Context) ([]v1alpha1.AdminNetworkPolicy, error) {
+	return m.AdminNetworkPolicies, m.AdminNetworkPolicyError
+}
+
+func (k *MockKubernetes) CreateAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.AdminNetworkPolicy) (*v1alpha1.AdminNetworkPolicy, error) {
+	return nil, ErrNotImplemented
+}
+
+func (k *MockKubernetes) UpdateAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.AdminNetworkPolicy) (*v1alpha1.AdminNetworkPolicy, error) {
+	return nil, ErrNotImplemented
+}
+
+func (k *MockKubernetes) DeleteAdminNetworkPolicy(ctx context.Context, name string) error {
+	//TODO: implement
+	return ErrNotImplemented
+}
+
+func (m *MockKubernetes) GetBaselineAdminNetworkPolicy(ctx context.Context) (*v1alpha1.BaselineAdminNetworkPolicy, error) {
+	return m.BaselineNetworkPolicy, m.BaseAdminNetworkPolicyError
+}
+
+func (k *MockKubernetes) CreateBaselineAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.BaselineAdminNetworkPolicy) (*v1alpha1.BaselineAdminNetworkPolicy, error) {
+	return nil, ErrNotImplemented
+}
+
+func (k *MockKubernetes) UpdateBaselineAdminNetworkPolicy(ctx context.Context, policy *v1alpha1.BaselineAdminNetworkPolicy) (*v1alpha1.BaselineAdminNetworkPolicy, error) {
+	return nil, ErrNotImplemented
+}
+
+func (k *MockKubernetes) DeleteBaselineAdminNetworkPolicy(ctx context.Context, name string) error {
+	//TODO: implement
+	return ErrNotImplemented
+}