Merge pull request #218 from tremes/statefulsets

Gather StatefulSet configs from default & openshift namespaces

Author: openshift-merge-robot

docs/insights-archive-sample/config/statefulsets/openshift-monitoring/prometheus-k8s.json

@@ -14,6 +14,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
+	appsclient "k8s.io/client-go/kubernetes/typed/apps/v1"
 	policyclient "k8s.io/client-go/kubernetes/typed/policy/v1beta1"
 	"k8s.io/client-go/pkg/version"
 	"k8s.io/client-go/rest"
@@ -139,6 +140,10 @@ func (s *Support) Run(ctx context.Context, controller *controllercmd.ControllerC
 		return err
 	}
 
+	appsClient, err := appsclient.NewForConfig(gatherKubeConfig)
+	if err != nil {
+		return err
+	}
 	// ensure the insight snapshot directory exists
 	if _, err := os.Stat(s.StoragePath); err != nil && os.IsNotExist(err) {
 		if err := os.MkdirAll(s.StoragePath, 0777); err != nil {
@@ -161,7 +166,7 @@ func (s *Support) Run(ctx context.Context, controller *controllercmd.ControllerC
 
 	// the gatherers periodically check the state of the cluster and report any
 	// config to the recorder
-	configPeriodic := clusterconfig.New(gatherConfigClient, gatherKubeClient.CoreV1(), gatherKubeClient.CertificatesV1beta1(), metricsClient, registryClient.ImageregistryV1(), crdClient, gatherNetworkClient, dynamicClient, gatherPolicyClient)
+	configPeriodic := clusterconfig.New(gatherConfigClient, gatherKubeClient.CoreV1(), gatherKubeClient.CertificatesV1beta1(), metricsClient, registryClient.ImageregistryV1(), crdClient, gatherNetworkClient, dynamicClient, gatherPolicyClient, appsClient)
 	periodic := periodic.New(configObserver, recorder, map[string]gather.Interface{
 		"config": configPeriodic,
 	})

pkg/controller/operator.go

@@ -13,7 +13,6 @@ type Interface interface {
 
 type Operation string
 
-
 const (
 	// DownloadingReport specific flag for Smart Proxy report downloading process.
 	DownloadingReport Operation = "DownloadingReport"

pkg/controllerstatus/controllerstatus.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/dynamic"
 	kubescheme "k8s.io/client-go/kubernetes/scheme"
+	appsclient "k8s.io/client-go/kubernetes/typed/apps/v1"
 	certificatesv1beta1 "k8s.io/client-go/kubernetes/typed/certificates/v1beta1"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	policyclient "k8s.io/client-go/kubernetes/typed/policy/v1beta1"
@@ -42,6 +43,7 @@ import (
 	configv1client "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
 	imageregistryv1 "github.com/openshift/client-go/imageregistry/clientset/versioned/typed/imageregistry/v1"
 	networkv1client "github.com/openshift/client-go/network/clientset/versioned/typed/network/v1"
+	appsv1 "k8s.io/api/apps/v1"
 	_ "k8s.io/apimachinery/pkg/runtime/serializer/yaml"
 
 	"github.com/openshift/insights-operator/pkg/record"
@@ -81,6 +83,7 @@ var (
 	openshiftSerializer     = openshiftscheme.Codecs.LegacyCodec(configv1.SchemeGroupVersion)
 	kubeSerializer          = kubescheme.Codecs.LegacyCodec(corev1.SchemeGroupVersion)
 	policyV1Beta1Serializer = kubescheme.Codecs.LegacyCodec(policyv1beta1.SchemeGroupVersion)
+	appsV1Serializer        = kubescheme.Codecs.LegacyCodec(appsv1.SchemeGroupVersion)
 	// maxEventTimeInterval represents the "only keep events that are maximum 1h old"
 	// TODO: make this dynamic like the reporting window based on configured interval
 	maxEventTimeInterval = 1 * time.Hour
@@ -120,14 +123,15 @@ type Gatherer struct {
 	registryClient imageregistryv1.ImageregistryV1Interface
 	crdClient      apixv1beta1client.ApiextensionsV1beta1Interface
 	policyClient   policyclient.PolicyV1beta1Interface
+	appsClient     appsclient.AppsV1Interface
 	lock           sync.Mutex
 	lastVersion    *configv1.ClusterVersion
 }
 
 // New creates new Gatherer
 func New(client configv1client.ConfigV1Interface, coreClient corev1client.CoreV1Interface, certClient certificatesv1beta1.CertificatesV1beta1Interface, metricsClient rest.Interface,
 	registryClient imageregistryv1.ImageregistryV1Interface, crdClient apixv1beta1client.ApiextensionsV1beta1Interface, networkClient networkv1client.NetworkV1Interface,
-	dynamicClient dynamic.Interface, policyClient policyclient.PolicyV1beta1Interface) *Gatherer {
+	dynamicClient dynamic.Interface, policyClient policyclient.PolicyV1beta1Interface, appsclient appsclient.AppsV1Interface) *Gatherer {
 	return &Gatherer{
 		client:         client,
 		coreClient:     coreClient,
@@ -138,6 +142,7 @@ func New(client configv1client.ConfigV1Interface, coreClient corev1client.CoreV1
 		networkClient:  networkClient,
 		dynamicClient:  dynamicClient,
 		policyClient:   policyClient,
+		appsClient:     appsclient,
 	}
 }
 
@@ -172,6 +177,7 @@ func (i *Gatherer) Gather(ctx context.Context, recorder record.Interface) error
 		GatherServiceAccounts(i),
 		GatherMachineConfigPool(i),
 		GatherContainerRuntimeConfig(i),
+		GatherStatefulSets(i),
 	)
 }
 
@@ -1054,6 +1060,47 @@ func GatherContainerRuntimeConfig(i *Gatherer) func() ([]record.Record, []error)
 	}
 }
 
+//GatherStatefulSets collects StatefulSet configs from default namespaces
+//
+// The Kubernetes API https://github.com/kubernetes/api/blob/master/apps/v1/types.go
+// Response see https://docs.openshift.com/container-platform/4.5/rest_api/workloads_apis/statefulset-apps-v1.html#statefulset-apps-v1
+//
+// Location in archive: config/statefulsets/
+func GatherStatefulSets(i *Gatherer) func() ([]record.Record, []error) {
+	return func() ([]record.Record, []error) {
+		namespaces, err := getAllNamespaces(i)
+		if errors.IsNotFound(err) {
+			return nil, nil
+		}
+		if err != nil {
+			return nil, []error{err}
+		}
+
+		osNamespaces := defaultNamespaces
+		for _, item := range namespaces.Items {
+			if strings.HasPrefix(item.Name, "openshift") {
+				osNamespaces = append(osNamespaces, item.Name)
+			}
+		}
+		records := []record.Record{}
+		for _, namespace := range osNamespaces {
+			sets, err := i.appsClient.StatefulSets(namespace).List(i.ctx, metav1.ListOptions{})
+			if err != nil {
+				klog.V(2).Infof("Unable to read StatefulSets in namespace %s error %s", namespace, err)
+				continue
+			}
+
+			for _, i := range sets.Items {
+				records = append(records, record.Record{
+					Name: fmt.Sprintf("config/statefulsets/%s/%s", namespace, i.GetName()),
+					Item: StatefulSetAnonymizer{&i},
+				})
+			}
+		}
+		return records, nil
+	}
+}
+
 func (i *Gatherer) gatherNamespaceEvents(namespace string) ([]record.Record, []error) {
 	// do not accidentally collect events for non-openshift namespace
 	if !strings.HasPrefix(namespace, "openshift-") {
@@ -1837,3 +1884,16 @@ func (a ServiceAccountsMarshaller) Marshal(_ context.Context) ([]byte, error) {
 func (a ServiceAccountsMarshaller) GetExtension() string {
 	return "json"
 }
+
+// StatefulSetAnonymizer implements StatefulSet serialization without anonymization
+type StatefulSetAnonymizer struct{ *appsv1.StatefulSet }
+
+// Marshal implements StatefulSet serialization
+func (a StatefulSetAnonymizer) Marshal(_ context.Context) ([]byte, error) {
+	return runtime.Encode(appsV1Serializer, a.StatefulSet)
+}
+
+// GetExtension returns extension for StatefulSet object
+func (a StatefulSetAnonymizer) GetExtension() string {
+	return "json"
+}

pkg/gather/clusterconfig/clusterconfig.go

@@ -11,6 +11,7 @@ import (
 
 	imageregistryv1 "github.com/openshift/api/imageregistry/v1"
 	networkv1 "github.com/openshift/api/network/v1"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
@@ -873,6 +874,43 @@ func TestGatherServiceAccounts(t *testing.T) {
 	}
 }
 
+func TestGatherStatefulSet(t *testing.T) {
+	testSet := appsv1.StatefulSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-statefulset",
+			Namespace: "openshift-test",
+		},
+	}
+	client := kubefake.NewSimpleClientset()
+	_, err := client.CoreV1().Namespaces().Create(context.Background(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "openshift-test"}}, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal("unable to create fake namespace", err)
+	}
+	_, err = client.AppsV1().StatefulSets("openshift-test").Create(context.Background(), &testSet, metav1.CreateOptions{})
+	if err != nil {
+		t.Fatal("unable to create fake statefulset", err)
+	}
+
+	gatherer := &Gatherer{ctx: context.Background(), coreClient: client.CoreV1(), appsClient: client.AppsV1()}
+
+	records, errs := GatherStatefulSets(gatherer)()
+	if len(errs) > 0 {
+		t.Errorf("unexpected errors: %#v", errs)
+		return
+	}
+
+	item, err := records[0].Item.Marshal(context.TODO())
+	var gatheredStatefulSet appsv1.StatefulSet
+	_, _, err = appsV1Serializer.Decode(item, nil, &gatheredStatefulSet)
+	if err != nil {
+		t.Fatalf("failed to decode object: %v", err)
+	}
+	if gatheredStatefulSet.Name != "test-statefulset" {
+		t.Fatalf("unexpected statefulset name %s", gatheredStatefulSet.Name)
+	}
+
+}
+
 func ExampleGatherMostRecentMetrics_Test() {
 	b, err := ExampleMostRecentMetrics()
 	if err != nil {

pkg/gather/clusterconfig/clusterconfig_test.go

@@ -42,7 +42,7 @@ func TestPullSecretExists(t *testing.T) {
 }
 
 func TestIsIOHealthy(t *testing.T) {
-	checkPodsLogs(t,  `The operator is healthy`)
+	checkPodsLogs(t, `The operator is healthy`)
 }
 
 // Check if opt-in/opt-out works

test/integration/basic_test.go

@@ -28,7 +28,7 @@ func TestUploadNotDelayedAfterStart(t *testing.T) {
 	time1 := logLineTime(t, `Reporting status periodically to .* every`)
 	time2 := logLineTime(t, `Successfully reported id=`)
 	delay := time2.Sub(time1)
-	allowedDelay := 3*time.Minute
+	allowedDelay := 3 * time.Minute
 	t.Logf("Archive upload delay was %d seconds", delay/time.Second)
 	if delay > allowedDelay && delay < time.Hour*24-allowedDelay {
 		t.Fatal("Upload after start took too much time")