diff --git a/pkg/ddc/cache/engine/cm.go b/pkg/ddc/cache/engine/cm.go index 8218f09ff5a..15c8ddcdca8 100644 --- a/pkg/ddc/cache/engine/cm.go +++ b/pkg/ddc/cache/engine/cm.go @@ -104,12 +104,10 @@ func (e *CacheEngine) generateRuntimeConfigData(ctx context.Context, runtime *da if err != nil { return nil, err } - runtimeClass, err := e.getRuntimeClass(runtime.Spec.RuntimeClassName) if err != nil { return nil, err } - var mounts []common.MountConfig for _, m := range dataset.Spec.Mounts { mountCg := common.MountConfig{ @@ -139,7 +137,7 @@ func (e *CacheEngine) generateRuntimeConfigData(ctx context.Context, runtime *da config.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany} } - if !runtime.Spec.Master.Disabled { + if runtimeClass.Topology.Master != nil && !runtime.Spec.Master.Disabled { config.Master = &common.CacheRuntimeComponentConfig{ Enabled: true, Name: common.GetCacheComponentName(e.name, common.ComponentTypeMaster), @@ -153,7 +151,7 @@ func (e *CacheEngine) generateRuntimeConfigData(ctx context.Context, runtime *da } } } - if !runtime.Spec.Worker.Disabled { + if runtimeClass.Topology.Worker != nil && !runtime.Spec.Worker.Disabled { config.Worker = &common.CacheRuntimeComponentConfig{ Enabled: true, Name: common.GetCacheComponentName(e.name, common.ComponentTypeWorker), @@ -169,7 +167,7 @@ func (e *CacheEngine) generateRuntimeConfigData(ctx context.Context, runtime *da // Extract tiered store configuration for worker config.Worker.TieredStoreLevels = e.extractTieredStoreLevels(&runtime.Spec.Worker.TieredStore) } - if !runtime.Spec.Client.Disabled { + if runtimeClass.Topology.Client != nil && !runtime.Spec.Client.Disabled { config.Client = &common.CacheRuntimeComponentConfig{ Enabled: true, Name: common.GetCacheComponentName(e.name, common.ComponentTypeClient), diff --git a/pkg/ddc/cache/engine/cm_test.go b/pkg/ddc/cache/engine/cm_test.go index 48d3c21a468..d94ce5d0ae4 100644 --- a/pkg/ddc/cache/engine/cm_test.go +++ b/pkg/ddc/cache/engine/cm_test.go @@ -18,6 +18,7 @@ package engine import ( "context" + "strings" "testing" datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1" @@ -207,6 +208,11 @@ func newCacheRuntimeForConfigMapTest() *datav1alpha1.CacheRuntime { func newCacheRuntimeClassForConfigMapTest() *datav1alpha1.CacheRuntimeClass { return &datav1alpha1.CacheRuntimeClass{ ObjectMeta: metav1.ObjectMeta{Name: "test-class"}, + Topology: &datav1alpha1.RuntimeTopology{ + Master: &datav1alpha1.RuntimeComponentDefinition{}, + Worker: &datav1alpha1.RuntimeComponentDefinition{}, + Client: &datav1alpha1.RuntimeComponentDefinition{}, + }, } } @@ -228,3 +234,103 @@ func newDatasetForConfigMapTest() *datav1alpha1.Dataset { Status: datav1alpha1.DatasetStatus{Runtimes: []datav1alpha1.Runtime{{Name: "demo", Type: common.CacheRuntime}}}, } } +func TestGenerateRuntimeConfigDataWithMissingComponentTopology(t *testing.T) { + testCases := map[string]struct { + topology *datav1alpha1.RuntimeTopology + enable func(runtime *datav1alpha1.CacheRuntime) + }{ + "master topology undefined": { + topology: &datav1alpha1.RuntimeTopology{ + Worker: &datav1alpha1.RuntimeComponentDefinition{}, + Client: &datav1alpha1.RuntimeComponentDefinition{}, + }, + enable: func(r *datav1alpha1.CacheRuntime) { r.Spec.Master.Disabled = false }, + }, + "worker topology undefined": { + topology: &datav1alpha1.RuntimeTopology{ + Master: &datav1alpha1.RuntimeComponentDefinition{}, + Client: &datav1alpha1.RuntimeComponentDefinition{}, + }, + enable: func(r *datav1alpha1.CacheRuntime) { r.Spec.Worker.Disabled = false }, + }, + "client topology undefined": { + topology: &datav1alpha1.RuntimeTopology{ + Master: &datav1alpha1.RuntimeComponentDefinition{}, + Worker: &datav1alpha1.RuntimeComponentDefinition{}, + }, + enable: func(r *datav1alpha1.CacheRuntime) { r.Spec.Client.Disabled = false }, + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + scheme := newCacheEngineTestScheme(t) + runtimeObj := newCacheRuntimeForConfigMapTest() + tc.enable(runtimeObj) + runtimeClass := newCacheRuntimeClassForConfigMapTest() + runtimeClass.Topology = tc.topology + dataset := newDatasetForConfigMapTest() + baseClient := fake.NewFakeClientWithScheme(scheme, runtimeObj, runtimeClass, dataset) + engine := &CacheEngine{Client: baseClient, name: "demo", namespace: "default"} + + if _, err := engine.generateRuntimeConfigData(context.Background(), runtimeObj); err != nil { + t.Fatalf("expected no error, got %v", err) + } + }) + } +} +func TestGenerateRuntimeConfigDataWithoutAnyComponent(t *testing.T) { + testCases := map[string]struct { + topology *datav1alpha1.RuntimeTopology + }{ + "topology is nil": {topology: nil}, + "topology declares no component": {topology: &datav1alpha1.RuntimeTopology{}}, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + scheme := newCacheEngineTestScheme(t) + runtimeObj := newCacheRuntimeForConfigMapTest() + runtimeClass := newCacheRuntimeClassForConfigMapTest() + runtimeClass.Topology = tc.topology + dataset := newDatasetForConfigMapTest() + baseClient := fake.NewFakeClientWithScheme(scheme, runtimeObj, runtimeClass, dataset) + engine := &CacheEngine{Client: baseClient, name: "demo", namespace: "default"} + + _, err := engine.generateRuntimeConfigData(context.Background(), runtimeObj) + if err == nil { + t.Fatal("expected error, got nil") + } + if !strings.Contains(err.Error(), "at least one component should be defined") { + t.Fatalf("unexpected error message: %v", err) + } + }) + } +} +func TestGenerateDataLoadValueFileWithNilTopology(t *testing.T) { + scheme := newCacheEngineTestScheme(t) + runtimeObj := newCacheRuntimeForConfigMapTest() + runtimeClass := &datav1alpha1.CacheRuntimeClass{ + ObjectMeta: metav1.ObjectMeta{Name: "test-class"}, + Topology: nil, + DataOperationSpecs: []datav1alpha1.DataOperationSpec{ + {Name: "DataLoad", Command: []string{"/usr/local/bin/dataload"}}, + }, + } + dataset := newDatasetForConfigMapTest() + dataload := &datav1alpha1.DataLoad{ + ObjectMeta: metav1.ObjectMeta{Name: "test-dataload", Namespace: "default"}, + Spec: datav1alpha1.DataLoadSpec{ + Dataset: datav1alpha1.TargetDataset{Name: "demo", Namespace: "default"}, + }, + } + + fakeClient := fake.NewFakeClientWithScheme(scheme, runtimeObj, runtimeClass, dataset, dataload) + engine := &CacheEngine{Client: fakeClient, name: "demo", namespace: "default"} + ctx := cruntime.ReconcileRequestContext{Client: fakeClient} + + _, err := engine.generateDataLoadValueFile(ctx, dataload) + if err == nil { + t.Fatal("expected error when topology is nil, got nil") + } +} diff --git a/pkg/ddc/cache/engine/runtime.go b/pkg/ddc/cache/engine/runtime.go index 4a45b25d9a7..55b1e642713 100644 --- a/pkg/ddc/cache/engine/runtime.go +++ b/pkg/ddc/cache/engine/runtime.go @@ -70,7 +70,6 @@ func (e *CacheEngine) getRuntime() (*datav1alpha1.CacheRuntime, error) { if err := e.Get(context.TODO(), key, &runtime); err != nil { return nil, err } - return &runtime, nil } @@ -82,7 +81,9 @@ func (e *CacheEngine) getRuntimeClass(runtimeClassName string) (*datav1alpha1.Ca if err := e.Get(context.TODO(), key, &runtimeClass); err != nil { return nil, err } - + if err := validateRuntimeClassTopology(&runtimeClass); err != nil { + return nil, err + } return &runtimeClass, nil } diff --git a/pkg/ddc/cache/engine/transform.go b/pkg/ddc/cache/engine/transform.go index e2277dd30dd..57251080a3f 100644 --- a/pkg/ddc/cache/engine/transform.go +++ b/pkg/ddc/cache/engine/transform.go @@ -17,7 +17,6 @@ package engine import ( - "fmt" "time" datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1" @@ -48,10 +47,8 @@ type RuntimeConfigVolumeConfig struct { } func (e *CacheEngine) transform(dataset *datav1alpha1.Dataset, runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (*common.CacheRuntimeValue, error) { - - if runtimeClass.Topology == nil || - (runtimeClass.Topology.Master == nil && runtimeClass.Topology.Worker == nil && runtimeClass.Topology.Client == nil) { - return nil, fmt.Errorf("at least one component should be defined in runtimeClass") + if err := validateRuntimeClassTopology(runtimeClass); err != nil { + return nil, err } defer utils.TimeTrack(time.Now(), "CacheRuntime.transform", "name", runtime.Name) @@ -83,9 +80,8 @@ func (e *CacheEngine) transform(dataset *datav1alpha1.Dataset, runtime *datav1al // getRuntimeStatusValue extracts minimal component status information from runtimeClass and runtime spec // This is a lightweight alternative to transform() when only status update is needed func (e *CacheEngine) getRuntimeStatusValue(runtime *datav1alpha1.CacheRuntime, runtimeClass *datav1alpha1.CacheRuntimeClass) (*common.CacheRuntimeStatusValue, error) { - if runtimeClass.Topology == nil || - (runtimeClass.Topology.Master == nil && runtimeClass.Topology.Worker == nil && runtimeClass.Topology.Client == nil) { - return nil, fmt.Errorf("at least one component should be defined in runtimeClass") + if err := validateRuntimeClassTopology(runtimeClass); err != nil { + return nil, err } statusValue := &common.CacheRuntimeStatusValue{} diff --git a/pkg/ddc/cache/engine/validate.go b/pkg/ddc/cache/engine/validate.go index 6721782bc0c..143aac4b391 100644 --- a/pkg/ddc/cache/engine/validate.go +++ b/pkg/ddc/cache/engine/validate.go @@ -17,9 +17,20 @@ package engine import ( + "fmt" + + datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1" "github.com/fluid-cloudnative/fluid/pkg/runtime" ) func (e *CacheEngine) Validate(runtime.ReconcileRequestContext) (err error) { return nil } + +func validateRuntimeClassTopology(runtimeClass *datav1alpha1.CacheRuntimeClass) error { + if runtimeClass.Topology == nil || + (runtimeClass.Topology.Master == nil && runtimeClass.Topology.Worker == nil && runtimeClass.Topology.Client == nil) { + return fmt.Errorf("at least one component should be defined in runtimeClass %s", runtimeClass.Name) + } + return nil +}