WIP: LOG-4445: Forwarder must use hosted cluster_id on HCP cluster.

internal/clusterinfo/_clusterinfo_static/clusterinfo_static.go

@@ -0,0 +1,51 @@
+package clusterinfo
+
+import (
+	"context"
+	configv1 "github.com/openshift/api/config/v1"
+	hypershiftv1beta1 "github.com/openshift/hypershift/api/hypershift/v1beta1"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func init() {
+	hypershiftv1beta1.AddToScheme(scheme.Scheme)
+	configv1.AddToScheme(scheme.Scheme)
+}
+
+func init() {
+	runtime.Must(hypershiftv1beta1.AddToScheme(scheme.Scheme))
+	runtime.Must(configv1.AddToScheme(scheme.Scheme))
+}
+
+// ClusterInfo is global information about where the ClusterLogForwarder is running.
+type ClusterInfo struct {
+	Version string // Version of the cluster.
+	ID      string // Unique identifier of the cluster.
+}
+
+// GetClusterInfo gets cluster info for the cluster we are running in.
+//
+// If namespace contains a HostedControlPlane then return info for the *guest* cluster, not the host cluster.
+// We assume in this case that CLF is running on behalf of the guest cluster to collect API audit logs.
+func Get(ctx context.Context, c client.Reader, namespace string) (ClusterInfo, error) {
+	// Use HCP info if exactly one HCP is present in the namespace.
+	hcps := &hypershiftv1beta1.HostedControlPlaneList{}
+	err := c.List(context.Background(), hcps, client.InNamespace(namespace))
+	if err == nil && len(hcps.Items) == 1 {
+		return ClusterInfo{
+			Version: hcps.Items[0].Status.VersionStatus.Desired.Version,
+			ID:      hcps.Items[0].Spec.ClusterID,
+		}, nil
+	}
+	// Use standalone ClusterVersion info.
+	cv := &configv1.ClusterVersion{}
+	if err := c.Get(ctx, client.ObjectKey{Name: "version"}, cv); err != nil {
+		return ClusterInfo{}, err
+	}
+	return ClusterInfo{
+		Version: cv.Spec.DesiredUpdate.Version,
+		ID:      string(cv.Spec.ClusterID),
+	}, nil
+}

internal/clusterinfo/_clusterinfo_static/clusterinfo_static_test.go

@@ -0,0 +1,54 @@
+package clusterinfo
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/ginkgo/extensions/table"
+	. "github.com/onsi/gomega"
+	configv1 "github.com/openshift/api/config/v1"
+	hypershiftv1beta1 "github.com/openshift/hypershift/api/hypershift/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func init() {
+	hypershiftv1beta1.AddToScheme(scheme.Scheme)
+}
+
+var _ = Describe("[internal][clusterinfo]", func() {
+	// All clusters have a ClusterVersion
+	clusterVersion := &configv1.ClusterVersion{
+		ObjectMeta: metav1.ObjectMeta{Name: "version"},
+		Spec: configv1.ClusterVersionSpec{
+			ClusterID:     "clusterVersion-id",
+			DesiredUpdate: &configv1.Update{Version: "clusterVersion-version"},
+		}}
+
+	// Each HCP management namespace has a HostedControlPlane
+	hcp := &hypershiftv1beta1.HostedControlPlane{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "testing", Name: "foobar"},
+		Spec: hypershiftv1beta1.HostedControlPlaneSpec{
+			ClusterID: "hypershift-id",
+		},
+		Status: hypershiftv1beta1.HostedControlPlaneStatus{
+			VersionStatus: &hypershiftv1beta1.ClusterVersionStatus{
+				Desired: configv1.Release{
+					Version: "hypershift-version",
+				},
+			},
+		},
+	}
+
+	DescribeTable("GetClusterInfo", func(id, version string, objects ...client.Object) {
+		c := fake.NewClientBuilder().WithObjects(objects...).Build()
+		info, err := Get(context.Background(), c, hcp.Namespace)
+		Expect(err).To(Succeed())
+		Expect(info).To(Equal(ClusterInfo{Version: version, ID: id}))
+	},
+		Entry("Standalone cluster", "clusterVersion-id", "clusterVersion-version", clusterVersion),
+		Entry("Hypershift cluster", "hypershift-id", "hypershift-version", clusterVersion, hcp),
+	)
+})

internal/clusterinfo/_clusterinfo_static/suite_test.go

@@ -0,0 +1,13 @@
+package clusterinfo
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestClusterInfo(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "[internal][clusterinfo] Suite")
+}

internal/clusterinfo/clusterinfo.go

@@ -0,0 +1,66 @@
+package clusterinfo
+
+import (
+	"context"
+
+	configv1 "github.com/openshift/api/config/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes/scheme"
+	client "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func init() {
+	runtime.Must(configv1.AddToScheme(scheme.Scheme))
+}
+
+// ClusterInfo is global information about where the ClusterLogForwarder is running.
+type ClusterInfo struct {
+	Version string // Version of the cluster.
+	ID      string // Unique identifier of the cluster.
+}
+
+// GetClusterInfo gets cluster info for the cluster we are running in.
+//
+// If is running in a Hosted Control Plane (hypershift management cluster)
+// this information describes the *guest* cluster, not the host cluster.
+// We assume in this case that CLF is running on behalf of the guest cluster to collect API audit logs.
+func Get(ctx context.Context, c client.Reader, namespace string) (*ClusterInfo, error) {
+	if hcp := getHCPInfo(ctx, c, namespace); hcp != nil { // Try HCP first
+		return hcp, nil
+	}
+	cv := &configv1.ClusterVersion{}
+	if err := c.Get(ctx, client.ObjectKey{Name: "version"}, cv); err != nil {
+		return nil, err
+	}
+	return &ClusterInfo{Version: cv.Spec.DesiredUpdate.Version, ID: string(cv.Spec.ClusterID)}, nil
+}
+
+var (
+	hcpGVK = schema.GroupVersionKind{
+		Group:   "hypershift.openshift.io",
+		Version: "v1beta1",
+		Kind:    "HostedControlPlane",
+	}
+)
+
+// getHCPInfo returns ClusterInfo from a hypershift control plane if found, nil otherwise.
+func getHCPInfo(ctx context.Context, c client.Reader, namespace string) *ClusterInfo {
+
+	// TODO: Using unstructured objects for HCP because of dependency problems with the hypershift API package.
+	// When https://issues.redhat.com/browse/HOSTEDCP-336 is fixed, switch to static approach.
+	// Code in ./_clusterinfo_static
+
+	l := &unstructured.UnstructuredList{}
+	l.SetGroupVersionKind(hcpGVK)
+	err := c.List(ctx, l, client.InNamespace(namespace))
+	if err != nil || len(l.Items) != 1 {
+		return nil
+	}
+	spec, _ := l.Items[0].Object["spec"].(map[string]any)
+	if spec == nil || spec["release"] == nil || spec["clusterID"] == nil {
+		return nil
+	}
+	return &ClusterInfo{Version: spec["release"].(string), ID: spec["clusterID"].(string)}
+}

internal/clusterinfo/clusterinfo_test.go

@@ -0,0 +1,60 @@
+package clusterinfo
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/ginkgo/extensions/table"
+	. "github.com/onsi/gomega"
+	configv1 "github.com/openshift/api/config/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func init() {
+	runtime.Must(configv1.AddToScheme(scheme.Scheme))
+}
+
+var _ = Describe("[internal][clusterinfo]", func() {
+
+	// All clusters have a ClusterVersion
+	cv := &configv1.ClusterVersion{
+		ObjectMeta: metav1.ObjectMeta{Name: "version"},
+		Spec: configv1.ClusterVersionSpec{
+			ClusterID:     "clusterVersion-id",
+			DesiredUpdate: &configv1.Update{Version: "clusterVersion-version"},
+		}}
+
+	// Each HCP management namespace has a HostedControlPlane
+	hcp := &unstructured.Unstructured{}
+	hcp.Object = map[string]any{
+		"spec": map[string]any{
+			"clusterID": "hypershift-id",
+			"release":   "hypershift-version",
+		},
+		"status": map[string]any{
+			"versionStatus": map[string]any{
+				"desired": map[string]any{ //configv1.Release{
+					"version": "hypershift-version",
+				},
+			},
+		},
+	}
+	hcp.SetGroupVersionKind(hcpGVK)
+	hcp.SetName("foobar")
+	hcp.SetNamespace("testing")
+
+	DescribeTable("Get", func(id, version string, objs ...client.Object) {
+		c := fake.NewClientBuilder().WithObjects(objs...).Build()
+		info, err := Get(context.Background(), c, hcp.GetNamespace())
+		Expect(err).To(Succeed())
+		Expect(info).To(Equal(&ClusterInfo{Version: version, ID: id}))
+	},
+		Entry("Uses hostedControlPlane in hypershift cluster", "hypershift-id", "hypershift-version", cv, hcp),
+		Entry("uses clusterversion in standalone cluster", "clusterVersion-id", "clusterVersion-version", cv),
+	)
+})

internal/clusterinfo/suite_test.go

@@ -0,0 +1,13 @@
+package clusterinfo
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+func TestV1Logging(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "[internal][clusterinfo] Suite")
+}

main.go

@@ -8,6 +8,7 @@ import (
 	"runtime"
 	"time"
 
+	"github.com/openshift/cluster-logging-operator/internal/clusterinfo"
 	"github.com/openshift/cluster-logging-operator/internal/metrics/dashboard"
 	"github.com/openshift/cluster-logging-operator/internal/metrics/telemetry"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -116,13 +117,11 @@ func main() {
 		}
 	}()
 
-	clusterVersion, err := getClusterVersion(mgr.GetAPIReader())
+	info, err := clusterinfo.Get(context.Background(), mgr.GetAPIReader(), getOpenshiftNS())
 	if err != nil {
 		log.Error(err, "unable to retrieve the clusterID")
 		os.Exit(1)
 	}
-	clusterID := string(clusterVersion.Spec.ClusterID)
-
 	migrateManifestResources(mgr.GetClient())
 
 	log.Info("Registering Components.")
@@ -132,8 +131,8 @@ func main() {
 		Reader:         mgr.GetAPIReader(),
 		Scheme:         mgr.GetScheme(),
 		Recorder:       mgr.GetEventRecorderFor("clusterlogging-controller"),
-		ClusterVersion: clusterVersion.Status.Desired.Version,
-		ClusterID:      clusterID,
+		ClusterVersion: info.Version,
+		ClusterID:      info.ID,
 	}).SetupWithManager(mgr); err != nil {
 		log.Error(err, "unable to create controller", "controller", "ClusterLogForwarder")
 		telemetry.Data.CLInfo.Set("healthStatus", UnHealthyStatus)
@@ -144,8 +143,8 @@ func main() {
 		Reader:         mgr.GetAPIReader(),
 		Scheme:         mgr.GetScheme(),
 		Recorder:       mgr.GetEventRecorderFor("clusterlogforwarder"),
-		ClusterVersion: clusterVersion.Status.Desired.Version,
-		ClusterID:      clusterID,
+		ClusterVersion: info.Version,
+		ClusterID:      info.ID,
 	}).SetupWithManager(mgr); err != nil {
 		log.Error(err, "unable to create controller", "controller", "ClusterLogging")
 		telemetry.Data.CLFInfo.Set("healthStatus", UnHealthyStatus)
@@ -157,8 +156,8 @@ func main() {
 		Client:         mgr.GetClient(),
 		Scheme:         mgr.GetScheme(),
 		Recorder:       mgr.GetEventRecorderFor("logfilemetricexporter"),
-		ClusterVersion: clusterVersion.Status.Desired.Version,
-		ClusterID:      clusterID,
+		ClusterVersion: info.Version,
+		ClusterID:      info.ID,
 	}).SetupWithManager(mgr); err != nil {
 		log.Error(err, "unable to create controller", "controller", "LogFileMetricExporter")
 		telemetry.Data.LFMEInfo.Set(telemetry.HealthStatus, UnHealthyStatus)
@@ -232,16 +231,6 @@ func getCLOVersion() (string, error) {
 	return cloversion, nil
 }
 
-// getClusterVersion retrieves the ID of the cluster
-func getClusterVersion(k8client client.Reader) (*configv1.ClusterVersion, error) {
-	clusterVersion := &configv1.ClusterVersion{}
-	key := client.ObjectKey{Name: "version"}
-	if err := k8client.Get(context.TODO(), key, clusterVersion); err != nil {
-		return nil, err
-	}
-	return clusterVersion, nil
-}
-
 func initLoggingResources(k8sClient client.Client, reader client.Reader) error {
 	// Create dashboard config map on CLO install
 	if err := dashboard.ReconcileDashboards(k8sClient, reader); err != nil {