diff options
Diffstat (limited to 'vnfs/DAaaS/microservices/remote-config-operator/pkg')
5 files changed, 172 insertions, 58 deletions
diff --git a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/prometheusremoteendpoint_types.go b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/prometheusremoteendpoint_types.go index 71b38dc2..bc633c1a 100644 --- a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/prometheusremoteendpoint_types.go +++ b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/prometheusremoteendpoint_types.go @@ -4,15 +4,10 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -type Metadata struct { - Name string `json:"Name"` - Labels []string `json:"Labels"` -} - // PrometheusRemoteEndpointSpec defines the desired state of PrometheusRemoteEndpoint // +k8s:openapi-gen=true type PrometheusRemoteEndpointSpec struct { - AdapterUrl string `json:"adapterUrl"` + AdapterURL string `json:"adapterURL"` FilterSelector *metav1.LabelSelector `json:"filterSelector,omitempty"` Type string `json:"type"` KafkaConfig string `json:"kafkaConfig,omitempty"` @@ -20,12 +15,16 @@ type PrometheusRemoteEndpointSpec struct { RemoteTimeout string `json:"remoteTimeout,omitempty"` } +// KafkaConfig - defines the desired remote kafka writer configurations type KafkaConfig struct { - BrokerUrl string `json:"brokerUrl"` - Group string `json:"group,omitempty"` - Topic string `json:"topic"` + BrokerURL string `json:"bootstrap.servers"` + Topic string `json:"topic"` + UsePartition bool `json:"usePartition"` + BatchMsgNum int `json:"batch.num.messages,omitempty` + Compression string `json:"compression.codec,omitempty` } +// QueueConfig - defines the prometheus remote write queue configurations type QueueConfig struct { BatchSendDeadline string `json:"batchSendDeadline,omitempty"` Capacity string `json:"capacity,omitempty"` @@ -39,10 +38,16 @@ type QueueConfig struct { // PrometheusRemoteEndpointStatus defines the observed state of PrometheusRemoteEndpoint // +k8s:openapi-gen=true +// +kubebuilder:subresource:status type PrometheusRemoteEndpointStatus struct { // INSERT ADDITIONAL STATUS FIELD - define observed state of cluster // Important: Run "operator-sdk generate k8s" to regenerate code after modifying this file // Add custom validation using kubebuilder tags: https://book-v1.book.kubebuilder.io/beyond_basics/generating_crd.html + // Status can be Error, Enabled + PrometheusInstance string `json:"prometheusInstance,omitempty"` + Status string `json:"status"` + KafkaWriterID string `json:"kafkaWriterID,omitempty"` + RemoteURL string `json:"remoteURL,omitempty"` } // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object diff --git a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/remotefilteraction_types.go b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/remotefilteraction_types.go index 18b3dc6d..4c09698c 100644 --- a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/remotefilteraction_types.go +++ b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/remotefilteraction_types.go @@ -7,8 +7,10 @@ import ( // RemoteFilterActionSpec defines the desired state of RemoteFilterAction // +k8s:openapi-gen=true type RemoteFilterActionSpec struct { - Action string `json:"action,omitempty"` - Regex string `json:"regex,omitempty"` + Action string `json:"action,omitempty"` + Regex string `json:"regex,omitempty"` + // SourceLabels are the labels of the each metric + // +listType=set SourceLabels []string `json:"sourceLabels,omitempty"` TargetLabel string `json:"targetLabel,omitempty"` Replacement string `json:"replacement,omitempty"` diff --git a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.deepcopy.go b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.deepcopy.go index 9fb59405..15aef9f7 100644 --- a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.deepcopy.go +++ b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.deepcopy.go @@ -26,27 +26,6 @@ func (in *KafkaConfig) DeepCopy() *KafkaConfig { } // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *Metadata) DeepCopyInto(out *Metadata) { - *out = *in - if in.Labels != nil { - in, out := &in.Labels, &out.Labels - *out = make([]string, len(*in)) - copy(*out, *in) - } - return -} - -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Metadata. -func (in *Metadata) DeepCopy() *Metadata { - if in == nil { - return nil - } - out := new(Metadata) - in.DeepCopyInto(out) - return out -} - -// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *PrometheusRemoteEndpoint) DeepCopyInto(out *PrometheusRemoteEndpoint) { *out = *in out.TypeMeta = in.TypeMeta @@ -78,7 +57,7 @@ func (in *PrometheusRemoteEndpoint) DeepCopyObject() runtime.Object { func (in *PrometheusRemoteEndpointList) DeepCopyInto(out *PrometheusRemoteEndpointList) { *out = *in out.TypeMeta = in.TypeMeta - out.ListMeta = in.ListMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) if in.Items != nil { in, out := &in.Items, &out.Items *out = make([]PrometheusRemoteEndpoint, len(*in)) @@ -193,7 +172,7 @@ func (in *RemoteFilterAction) DeepCopyObject() runtime.Object { func (in *RemoteFilterActionList) DeepCopyInto(out *RemoteFilterActionList) { *out = *in out.TypeMeta = in.TypeMeta - out.ListMeta = in.ListMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) if in.Items != nil { in, out := &in.Items, &out.Items *out = make([]RemoteFilterAction, len(*in)) diff --git a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.openapi.go b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.openapi.go index ca4c9511..3dba690e 100644 --- a/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.openapi.go +++ b/vnfs/DAaaS/microservices/remote-config-operator/pkg/apis/onap/v1alpha1/zz_generated.openapi.go @@ -71,7 +71,7 @@ func schema_pkg_apis_onap_v1alpha1_PrometheusRemoteEndpointSpec(ref common.Refer Description: "PrometheusRemoteEndpointSpec defines the desired state of PrometheusRemoteEndpoint", Type: []string{"object"}, Properties: map[string]spec.Schema{ - "adapterUrl": { + "adapterURL": { SchemaProps: spec.SchemaProps{ Type: []string{"string"}, Format: "", @@ -106,7 +106,7 @@ func schema_pkg_apis_onap_v1alpha1_PrometheusRemoteEndpointSpec(ref common.Refer }, }, }, - Required: []string{"adapterUrl", "type"}, + Required: []string{"adapterURL", "type"}, }, }, Dependencies: []string{ @@ -120,6 +120,34 @@ func schema_pkg_apis_onap_v1alpha1_PrometheusRemoteEndpointStatus(ref common.Ref SchemaProps: spec.SchemaProps{ Description: "PrometheusRemoteEndpointStatus defines the observed state of PrometheusRemoteEndpoint", Type: []string{"object"}, + Properties: map[string]spec.Schema{ + "prometheusInstance": { + SchemaProps: spec.SchemaProps{ + Description: "INSERT ADDITIONAL STATUS FIELD - define observed state of cluster Important: Run \"operator-sdk generate k8s\" to regenerate code after modifying this file Add custom validation using kubebuilder tags: https://book-v1.book.kubebuilder.io/beyond_basics/generating_crd.html Status can be Error, Enabled", + Type: []string{"string"}, + Format: "", + }, + }, + "status": { + SchemaProps: spec.SchemaProps{ + Type: []string{"string"}, + Format: "", + }, + }, + "kafkaWriterID": { + SchemaProps: spec.SchemaProps{ + Type: []string{"string"}, + Format: "", + }, + }, + "remoteURL": { + SchemaProps: spec.SchemaProps{ + Type: []string{"string"}, + Format: "", + }, + }, + }, + Required: []string{"status"}, }, }, } @@ -189,8 +217,14 @@ func schema_pkg_apis_onap_v1alpha1_RemoteFilterActionSpec(ref common.ReferenceCa }, }, "sourceLabels": { + VendorExtensible: spec.VendorExtensible{ + Extensions: spec.Extensions{ + "x-kubernetes-list-type": "set", + }, + }, SchemaProps: spec.SchemaProps{ - Type: []string{"array"}, + Description: "SourceLabels are the labels of the each metric", + Type: []string{"array"}, Items: &spec.SchemaOrArray{ Schema: &spec.Schema{ SchemaProps: spec.SchemaProps{ diff --git a/vnfs/DAaaS/microservices/remote-config-operator/pkg/controller/prometheusremoteendpoint/prometheusremoteendpoint_controller.go b/vnfs/DAaaS/microservices/remote-config-operator/pkg/controller/prometheusremoteendpoint/prometheusremoteendpoint_controller.go index 2d78a43f..914cb375 100644 --- a/vnfs/DAaaS/microservices/remote-config-operator/pkg/controller/prometheusremoteendpoint/prometheusremoteendpoint_controller.go +++ b/vnfs/DAaaS/microservices/remote-config-operator/pkg/controller/prometheusremoteendpoint/prometheusremoteendpoint_controller.go @@ -1,9 +1,12 @@ package prometheusremoteendpoint import ( + "bytes" "context" "encoding/json" + "net/http" "strconv" + "strings" onapv1alpha1 "remote-config-operator/pkg/apis/onap/v1alpha1" @@ -79,9 +82,8 @@ type ReconcilePrometheusRemoteEndpoint struct { scheme *runtime.Scheme } -// Reconcile reads that state of the cluster for a PrometheusRemoteEndpoint object and makes changes based on the state read -// and what is in the PrometheusRemoteEndpoint.Spec - +// Reconcile reads that state of the cluster for a PrometheusRemoteEndpoint object +// and makes changes based on the state read and what is in the PrometheusRemoteEndpoint.Spec // Note: // The Controller will requeue the Request to be processed again if the returned error is non-nil or // Result.Requeue is true, otherwise upon completion it will remove the work from the queue. @@ -105,8 +107,8 @@ func (r *ReconcilePrometheusRemoteEndpoint) Reconcile(request reconcile.Request) return reconcile.Result{}, err } - isBeingDeleted := checkDeletionTimestamp(reqLogger, instance) - if isBeingDeleted { + // Check if CR is being Deleted + if instance.GetDeletionTimestamp() != nil { //Delete Remote write if err := r.processDeletionRequest(reqLogger, instance); err != nil { reqLogger.Error(err, "Error processing deletion request") @@ -139,7 +141,8 @@ func (r *ReconcilePrometheusRemoteEndpoint) Reconcile(request reconcile.Request) func (r *ReconcilePrometheusRemoteEndpoint) processPatchRequest(reqLogger logr.Logger, instance *onapv1alpha1.PrometheusRemoteEndpoint) error { prom := &monitoringv1.Prometheus{} - if err1 := r.client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: instance.ObjectMeta.Labels["app"]}, prom); err1 != nil { + pName := instance.ObjectMeta.Labels["app"] + if err1 := r.client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: pName}, prom); err1 != nil { reqLogger.Error(err1, "Error getting prometheus") return err1 } @@ -148,13 +151,20 @@ func (r *ReconcilePrometheusRemoteEndpoint) processPatchRequest(reqLogger logr.L var patch []byte rws := prom.Spec.RemoteWrite - adapterURL := instance.Spec.AdapterUrl + remoteURL, id, err := getAdapterInfo(instance) + instanceKey := types.NamespacedName{Namespace: instance.Namespace, Name: instance.Name} + if err != nil { + reqLogger.Error(err, "Unable to get adapter url") + return err + } isUpdate := false for i, spec := range rws { - if spec.URL == instance.Spec.AdapterUrl { + // Update event - check the prometheus remote write Url against remoteURL in the Status + // to consider the case when URL itself is updated. + if spec.URL == instance.Status.RemoteURL { reqLogger.Info("Remote write already exists, updating it") - patch, _ = formPatch("replace", strconv.Itoa(i), adapterURL, instance, reqLogger) + patch, _ = formPatch("replace", strconv.Itoa(i), remoteURL, instance, reqLogger) isUpdate = true break } @@ -163,23 +173,21 @@ func (r *ReconcilePrometheusRemoteEndpoint) processPatchRequest(reqLogger logr.L if !isUpdate { reqLogger.Info("Remote write does not exist, creating one...") // rwsLength := len(rws) - patch, _ = formPatch("add", "-", adapterURL, instance, reqLogger) + patch, _ = formPatch("add", "-", remoteURL, instance, reqLogger) } patchErr := r.client.Patch(context.TODO(), prom, client.ConstantPatch(types.JSONPatchType, patch)) if patchErr != nil { reqLogger.Error(patchErr, "Unable to process patch to prometheus") + cleanUpExternalResources(instance) + r.updateStatus("Error", instanceKey, "", "", "") return patchErr } - reqLogger.Info("Patch merged") + r.updateStatus("Enabled", instanceKey, pName, remoteURL, id) + reqLogger.V(1).Info("Patch merged") return nil } -func checkDeletionTimestamp(reqlogger logr.Logger, instance *onapv1alpha1.PrometheusRemoteEndpoint) bool { - isMarkedForDeletion := instance.GetDeletionTimestamp() != nil - return isMarkedForDeletion -} - func (r *ReconcilePrometheusRemoteEndpoint) processDeletionRequest(reqLogger logr.Logger, instance *onapv1alpha1.PrometheusRemoteEndpoint) error { prom := &monitoringv1.Prometheus{} if err := r.client.Get(context.TODO(), types.NamespacedName{Namespace: instance.Namespace, Name: instance.ObjectMeta.Labels["app"]}, prom); err != nil { @@ -189,13 +197,17 @@ func (r *ReconcilePrometheusRemoteEndpoint) processDeletionRequest(reqLogger log reqLogger.Info("Found prometheus to update") var patch []byte - adapterURL := instance.Spec.AdapterUrl + remoteURL, _, err := getAdapterInfo(instance) + if err != nil { + reqLogger.Error(err, "Unable to get adapter info") + return err + } rws := prom.Spec.RemoteWrite for i, spec := range rws { - if spec.URL == instance.Spec.AdapterUrl { + if spec.URL == remoteURL { reqLogger.Info("Found remote write to be removed, removing it") - patch, _ = formPatch("remove", strconv.Itoa(i), adapterURL, instance, reqLogger) + patch, _ = formPatch("remove", strconv.Itoa(i), remoteURL, instance, reqLogger) break } } @@ -204,8 +216,8 @@ func (r *ReconcilePrometheusRemoteEndpoint) processDeletionRequest(reqLogger log reqLogger.Error(patchErr, "Unable to process patch to prometheus") return patchErr } - reqLogger.Info("Patch merged, remote write removed") - + reqLogger.V(1).Info("Patch merged, remote write removed") + cleanUpExternalResources(instance) //remove Finalizer after deletion if remoteconfigutils.Contains(instance.GetFinalizers(), remoteconfigutils.RemoteConfigFinalizer) { if err := removeFinalizer(reqLogger, instance); err != nil { @@ -252,3 +264,85 @@ func formPatch(method string, index string, adapterURL string, instance *onapv1a finalMergePatch := append(prependMergePatch, 93) return finalMergePatch, nil } + +func (r *ReconcilePrometheusRemoteEndpoint) updateStatus(status string, key types.NamespacedName, prom string, remoteURL string, kwid string) error { + // Fetch the CollectdGlobal instance + instance := &onapv1alpha1.PrometheusRemoteEndpoint{} + err := r.client.Get(context.TODO(), key, instance) + if err != nil { + return err + } + instance.Status.Status = status + instance.Status.PrometheusInstance = prom + instance.Status.KafkaWriterID = kwid + instance.Status.RemoteURL = remoteURL + err = r.client.Status().Update(context.TODO(), instance) + return err +} + +func cleanUpExternalResources(instance *onapv1alpha1.PrometheusRemoteEndpoint) { + if instance.Spec.Type == "kafka" { + deleteKafkaWriter(instance.Spec.AdapterURL + "/pkw/" + instance.Status.KafkaWriterID) + } +} + +func getAdapterInfo(instance *onapv1alpha1.PrometheusRemoteEndpoint) (remoteURL string, id string, err error) { + switch strings.ToLower(instance.Spec.Type) { + case "m3db": + return instance.Spec.AdapterURL + "/api/v1/prom/remote/write", "", nil + case "kafka": + kwid, err := getKafkaWriter(instance) + return instance.Spec.AdapterURL + "/pkw/" + kwid + "/receive", kwid, err + default: + return instance.Spec.AdapterURL, "", nil + } +} + +func deleteKafkaWriter(kwURL string) error { + client := &http.Client{} + req, err := http.NewRequest(http.MethodDelete, kwURL, nil) + if err != nil { + log.Error(err, "Failed to form delete Kafka Writer request") + return err + } + _, err = client.Do(req) + if err != nil { + log.Error(err, "Failed to delete Kafka Writer", "Kafka Writer", kwURL) + return err + } + return nil +} + +func getKafkaWriter(instance *onapv1alpha1.PrometheusRemoteEndpoint) (string, error) { + // TODO - check update events + if instance.Status.KafkaWriterID != "" { + return instance.Status.KafkaWriterID, nil + } + return createKafkaWriter(instance) +} + +func createKafkaWriter(instance *onapv1alpha1.PrometheusRemoteEndpoint) (string, error) { + + log.V(1).Info("Processing Kafka Remote Endpoint", "Kafka Writer Config", instance.Spec) + baseURL := instance.Spec.AdapterURL + kwc := instance.Spec.KafkaConfig + kwURL := baseURL + "/pkw" + + postBody, err := json.Marshal(kwc) + if err != nil { + log.Error(err, "JSON Marshalling error") + return "", err + } + + resp, err := http.Post(kwURL, "application/json", bytes.NewBuffer(postBody)) + if err != nil { + log.Error(err, "Failed to create Kafka Writer", "Kafka Writer", kwURL, "Kafka Writer Config", kwc) + return "", err + } + defer resp.Body.Close() + var kwid string + json.NewDecoder(resp.Body).Decode(&kwid) + log.Info("Kafka Writer created", "Kafka Writer Id", kwid) + + return kwid, err +} |