Merge pull request kubeagi#160 from kubeagi/bjwswang

fix: datasource check bugs

charts/arcadia/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: arcadia
 description: A Helm chart(KubeBB Component) for KubeAGI Arcadia
 type: application
-version: 0.1.10
+version: 0.1.11
 appVersion: "0.0.1"
 
 keywords:

charts/arcadia/charts/minio/templates/post-job.yaml

@@ -12,6 +12,7 @@ metadata:
   annotations:
     "helm.sh/hook": post-install,post-upgrade
     "helm.sh/hook-delete-policy": hook-succeeded,before-hook-creation
+    "helm.sh/hook-weight": "-1"    
     {{- with .Values.postJob.annotations }}
       {{- toYaml . | nindent 4 }}
     {{- end }}

charts/arcadia/templates/post-oss.yaml

@@ -5,6 +5,7 @@ metadata:
   namespace: {{ .Release.Namespace }}
   annotations:
     "helm.sh/hook": post-install,post-upgrade
+    "helm.sh/hook-weight": "1"
 spec:
   displayName: "内置系统数据源"
   description: "Arcadia 内置系统数据源"

controllers/datasource_controller.go

@@ -19,15 +19,19 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"reflect"
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	arcadiav1alpha1 "github.com/kubeagi/arcadia/api/v1alpha1"
@@ -99,7 +103,13 @@ func (r *DatasourceReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 // SetupWithManager sets up the controller with the Manager.
 func (r *DatasourceReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
-		For(&arcadiav1alpha1.Datasource{}).
+		For(&arcadiav1alpha1.Datasource{}, builder.WithPredicates(predicate.Funcs{
+			UpdateFunc: func(ue event.UpdateEvent) bool {
+				oldDatsource := ue.ObjectOld.(*arcadiav1alpha1.Datasource)
+				newDatasource := ue.ObjectNew.(*arcadiav1alpha1.Datasource)
+				return !reflect.DeepEqual(oldDatsource.Spec, newDatasource.Spec)
+			},
+		})).
 		Complete(r)
 }
 
@@ -173,27 +183,30 @@ func (r *DatasourceReconciler) Checkdatasource(ctx context.Context, logger logr.
 	return r.UpdateStatus(ctx, instance, nil)
 }
 
+// UpdateStatus uppon error
 func (r *DatasourceReconciler) UpdateStatus(ctx context.Context, instance *arcadiav1alpha1.Datasource, err error) error {
 	instanceCopy := instance.DeepCopy()
+	var newCondition arcadiav1alpha1.Condition
 	if err != nil {
-		// Set status to unavailable
-		instanceCopy.Status.SetConditions(arcadiav1alpha1.Condition{
+		// set condition to False
+		newCondition = arcadiav1alpha1.Condition{
 			Type:               arcadiav1alpha1.TypeReady,
 			Status:             corev1.ConditionFalse,
 			Reason:             arcadiav1alpha1.ReasonUnavailable,
 			Message:            err.Error(),
 			LastTransitionTime: metav1.Now(),
-		})
+		}
 	} else {
-		// Set status to available
-		instanceCopy.Status.SetConditions(arcadiav1alpha1.Condition{
+		// set condition to True
+		newCondition = arcadiav1alpha1.Condition{
 			Type:               arcadiav1alpha1.TypeReady,
 			Status:             corev1.ConditionTrue,
 			Reason:             arcadiav1alpha1.ReasonAvailable,
-			Message:            "health check success",
+			Message:            "Check Success",
 			LastTransitionTime: metav1.Now(),
 			LastSuccessfulTime: metav1.Now(),
-		})
+		}
 	}
+	instanceCopy.Status.SetConditions(newCondition)
 	return r.Client.Status().Update(ctx, instanceCopy)
 }

pkg/datasource/datasource.go

@@ -31,8 +31,6 @@ import (
 
 var (
 	ErrUnknowDatasourceType = errors.New("unknow datasource type")
-	ErrAccessPermission     = errors.New("dont have permission")
-	ErrAccessObject         = errors.New("access error object")
 )
 
 type Datasource interface {
@@ -75,6 +73,7 @@ func (local *Local) Check(ctx context.Context, options any) error {
 
 var _ Datasource = (*OSS)(nil)
 
+// OSS is a wrapper to object storage service
 type OSS struct {
 	*minio.Client
 }
@@ -90,7 +89,7 @@ func NewOSS(ctx context.Context, c client.Client, endpoint *v1alpha1.Endpoint) (
 			return nil, err
 		}
 		accessKeyID = string(secret.Data["rootUser"])
-		secretAccessKey = string(secret.Data["rootUser"])
+		secretAccessKey = string(secret.Data["rootPassword"])
 
 		// TODO: implement https(secure check)
 		// if !endpoint.Insecure {
@@ -108,30 +107,26 @@ func NewOSS(ctx context.Context, c client.Client, endpoint *v1alpha1.Endpoint) (
 	return &OSS{Client: mc}, nil
 }
 
+// Check oss agains info()
 func (oss *OSS) Check(ctx context.Context, info any) error {
 	if info == nil {
 		return nil
 	}
-	_, ok := info.(*v1alpha1.OSS)
+	ossInfo, ok := info.(*v1alpha1.OSS)
 	if !ok {
 		return errors.New("invalid check info for OSS")
 	}
 
-	if info.(*v1alpha1.OSS).Bucket != "" {
-		_, err := oss.Client.BucketExists(ctx, info.(*v1alpha1.OSS).Bucket)
+	if ossInfo.Bucket != "" {
+		_, err := oss.Client.BucketExists(ctx, ossInfo.Bucket)
 		if err != nil {
 			return err
 		}
 
-		if info.(*v1alpha1.OSS).Object != "" {
-			_, err := oss.Client.StatObject(ctx, info.(*v1alpha1.OSS).Bucket, info.(*v1alpha1.OSS).Object, minio.StatObjectOptions{})
+		if ossInfo.Object != "" {
+			_, err := oss.Client.StatObject(ctx, ossInfo.Bucket, ossInfo.Object, minio.StatObjectOptions{})
 			if err != nil {
-				switch minio.ToErrorResponse(err).Code {
-				case "AccessDenied":
-					return ErrAccessPermission
-				default:
-					return ErrAccessObject
-				}
+				return err
 			}
 		}
 	}