diff --git a/Dockerfile b/Dockerfile index d39c25a75..4741738a0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,13 +19,13 @@ COPY graphql-server/ graphql-server/ # Build RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a -o manager main.go -RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o go-bff-server graphql-server/go-server/main.go +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o graphql-server graphql-server/go-server/main.go # Use distroless as minimal base image to package the manager binary # Refer to https://github.com/GoogleContainerTools/distroless for more details FROM gcr.io/distroless/static:nonroot WORKDIR / COPY --from=builder /workspace/manager . -COPY --from=builder /workspace/go-bff-server . +COPY --from=builder /workspace/graphql-server . USER 65532:65532 ENTRYPOINT ["/manager"] diff --git a/PROJECT b/PROJECT index 194360c00..c5c2138b6 100644 --- a/PROJECT +++ b/PROJECT @@ -72,4 +72,22 @@ resources: kind: VersionedDataset path: github.com/kubeagi/arcadia/api/v1alpha1 version: v1alpha1 +- api: + crdVersion: v1 + namespaced: true + controller: true + domain: kubeagi.k8s.com.cn + group: arcadia + kind: Worker + path: github.com/kubeagi/arcadia/api/v1alpha1 + version: v1alpha1 +- api: + crdVersion: v1 + namespaced: true + controller: true + domain: kubeagi.k8s.com.cn + group: arcadia + kind: Model + path: github.com/kubeagi/arcadia/api/v1alpha1 + version: v1alpha1 version: "3" diff --git a/README.md b/README.md index 2900e6a7a..f664f2765 100644 --- a/README.md +++ b/README.md @@ -41,12 +41,12 @@ Our design and development in Arcadia design follows operator pattern which exte helm install --namespace arcadia --create-namespace arcadia arcadia/arcadia ``` -More conveniently,you can use [kubebb](https://github.com/kubebb) to install and upgrade arcadia automatically: -> Pre-requsities -> - [kubebb](https://kubebb.github.io/website/docs/quick-start/core_quickstart) -```shell -kubectl apply -f ./kubeagi.yaml -``` +> More conveniently,you can use [kubebb](https://github.com/kubebb) to install and upgrade arcadia automatically: + > Pre-requsities + > - [kubebb](https://kubebb.github.io/website/docs/quick-start/core_quickstart) +> ```shell +> kubectl apply -f ./kubeagi.yaml +> ``` ## CLI diff --git a/api/v1alpha1/datasource_types.go b/api/v1alpha1/datasource_types.go index 0fe805690..9d29c2b2e 100644 --- a/api/v1alpha1/datasource_types.go +++ b/api/v1alpha1/datasource_types.go @@ -44,6 +44,7 @@ type DatasourceSpec struct { // OSS defines info for object storage service as datasource type OSS struct { Bucket string `json:"bucket,omitempty"` + // Object must end with a slash "/" if it is a directory Object string `json:"object,omitempty"` } diff --git a/api/v1alpha1/model.go b/api/v1alpha1/model.go new file mode 100644 index 000000000..86a5a0138 --- /dev/null +++ b/api/v1alpha1/model.go @@ -0,0 +1,50 @@ +/* +Copyright 2023 KubeAGI. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package v1alpha1 + +import "fmt" + +const ( + // LabelModelType keeps the spec.type field + LabelModelType = Group + "/model-type" + LabelModelFullPath = Group + "/full-path" +) + +type ModelType string + +const ( + ModelTypeEmbedding ModelType = "embedding" + ModelTypeLLM ModelType = "llm" + ModelTypeUnknown ModelType = "unknown" +) + +func (model Model) ModelType() ModelType { + if model.Spec.Type == "" { + return ModelTypeUnknown + } + return model.Spec.Type +} + +// FullPath with bucket and object path +func (model Model) FullPath() string { + return fmt.Sprintf("%s/%s", model.Namespace, model.ObjectPath()) +} + +// ObjectPath is the path where model stored at in a bucket +func (model Model) ObjectPath() string { + return fmt.Sprintf("model/%s/", model.Name) +} diff --git a/api/v1alpha1/model_types.go b/api/v1alpha1/model_types.go new file mode 100644 index 000000000..11618d113 --- /dev/null +++ b/api/v1alpha1/model_types.go @@ -0,0 +1,75 @@ +/* +Copyright 2023 KubeAGI. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package v1alpha1 + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// EDIT THIS FILE! THIS IS SCAFFOLDING FOR YOU TO OWN! +// NOTE: json tags are required. Any new fields you add must have json tags for the fields to be serialized. + +// ModelSpec defines the desired state of Model +type ModelSpec struct { + // Creator defines dataset creator(AUTO-FILLED by webhook) + Creator string `json:"creator,omitempty"` + + // DisplayName defines dataset display name + DiplayName string `json:"displayName"` + + // Description defines datasource description + Description string `json:"description,omitempty"` + + // Type defines what kind of model this is + Type ModelType `json:"type,omitempty"` + + // TODO: extend model to utilize third party storage sources + // Source *TypedObjectReference `json:"source,omitempty"` + // // Path(relative to source) to the model files + // Path string `json:"path,omitempty"` +} + +// ModelStatus defines the observed state of Model +type ModelStatus struct { + // ConditionedStatus is the current status + ConditionedStatus `json:",inline"` +} + +//+kubebuilder:object:root=true +//+kubebuilder:subresource:status + +// Model is the Schema for the models API +type Model struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec ModelSpec `json:"spec,omitempty"` + Status ModelStatus `json:"status,omitempty"` +} + +//+kubebuilder:object:root=true + +// ModelList contains a list of Model +type ModelList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []Model `json:"items"` +} + +func init() { + SchemeBuilder.Register(&Model{}, &ModelList{}) +} diff --git a/api/v1alpha1/worker_types.go b/api/v1alpha1/worker_types.go new file mode 100644 index 000000000..9eb42d7c2 --- /dev/null +++ b/api/v1alpha1/worker_types.go @@ -0,0 +1,64 @@ +/* +Copyright 2023 KubeAGI. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package v1alpha1 + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// EDIT THIS FILE! THIS IS SCAFFOLDING FOR YOU TO OWN! +// NOTE: json tags are required. Any new fields you add must have json tags for the fields to be serialized. + +// WorkerSpec defines the desired state of Worker +type WorkerSpec struct { + // Creator defines dataset creator(AUTO-FILLED by webhook) + Creator string `json:"creator,omitempty"` + + // DisplayName defines dataset display name + DiplayName string `json:"displayName"` +} + +// WorkerStatus defines the observed state of Worker +type WorkerStatus struct { + // ConditionedStatus is the current status + ConditionedStatus `json:",inline"` +} + +//+kubebuilder:object:root=true +//+kubebuilder:subresource:status + +// Worker is the Schema for the workers API +type Worker struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec WorkerSpec `json:"spec,omitempty"` + Status WorkerStatus `json:"status,omitempty"` +} + +//+kubebuilder:object:root=true + +// WorkerList contains a list of Worker +type WorkerList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []Worker `json:"items"` +} + +func init() { + SchemeBuilder.Register(&Worker{}, &WorkerList{}) +} diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index d849be646..1f5343a92 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -579,6 +579,96 @@ func (in *LaboratoryStatus) DeepCopy() *LaboratoryStatus { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *Model) DeepCopyInto(out *Model) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + out.Spec = in.Spec + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Model. +func (in *Model) DeepCopy() *Model { + if in == nil { + return nil + } + out := new(Model) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *Model) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ModelList) DeepCopyInto(out *ModelList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]Model, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelList. +func (in *ModelList) DeepCopy() *ModelList { + if in == nil { + return nil + } + out := new(ModelList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ModelList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ModelSpec) DeepCopyInto(out *ModelSpec) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelSpec. +func (in *ModelSpec) DeepCopy() *ModelSpec { + if in == nil { + return nil + } + out := new(ModelSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ModelStatus) DeepCopyInto(out *ModelStatus) { + *out = *in + in.ConditionedStatus.DeepCopyInto(&out.ConditionedStatus) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ModelStatus. +func (in *ModelStatus) DeepCopy() *ModelStatus { + if in == nil { + return nil + } + out := new(ModelStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *OSS) DeepCopyInto(out *OSS) { *out = *in @@ -820,3 +910,93 @@ func (in *VersionedDatasetStatus) DeepCopy() *VersionedDatasetStatus { in.DeepCopyInto(out) return out } + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *Worker) DeepCopyInto(out *Worker) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + out.Spec = in.Spec + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Worker. +func (in *Worker) DeepCopy() *Worker { + if in == nil { + return nil + } + out := new(Worker) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *Worker) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *WorkerList) DeepCopyInto(out *WorkerList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]Worker, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerList. +func (in *WorkerList) DeepCopy() *WorkerList { + if in == nil { + return nil + } + out := new(WorkerList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *WorkerList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *WorkerSpec) DeepCopyInto(out *WorkerSpec) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerSpec. +func (in *WorkerSpec) DeepCopy() *WorkerSpec { + if in == nil { + return nil + } + out := new(WorkerSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *WorkerStatus) DeepCopyInto(out *WorkerStatus) { + *out = *in + in.ConditionedStatus.DeepCopyInto(&out.ConditionedStatus) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new WorkerStatus. +func (in *WorkerStatus) DeepCopy() *WorkerStatus { + if in == nil { + return nil + } + out := new(WorkerStatus) + in.DeepCopyInto(out) + return out +} diff --git a/charts/arcadia/Chart.yaml b/charts/arcadia/Chart.yaml index 7ae631fc9..d3f709d04 100644 --- a/charts/arcadia/Chart.yaml +++ b/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.15 +version: 0.1.16 appVersion: "0.0.1" keywords: diff --git a/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_datasources.yaml b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_datasources.yaml index 0f237d92e..2883a043d 100644 --- a/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_datasources.yaml +++ b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_datasources.yaml @@ -92,8 +92,11 @@ spec: bucket: type: string object: + description: Object must end with a slash "/" if it is a directory type: string type: object + required: + - displayName type: object status: description: DatasourceStatus defines the observed state of Datasource diff --git a/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_models.yaml b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_models.yaml new file mode 100644 index 000000000..50434d738 --- /dev/null +++ b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_models.yaml @@ -0,0 +1,99 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.9.2 + creationTimestamp: null + name: models.arcadia.kubeagi.k8s.com.cn +spec: + group: arcadia.kubeagi.k8s.com.cn + names: + kind: Model + listKind: ModelList + plural: models + singular: model + scope: Namespaced + versions: + - name: v1alpha1 + schema: + openAPIV3Schema: + description: Model is the Schema for the models API + properties: + apiVersion: + description: 'APIVersion defines the versioned schema of this representation + of an object. Servers should convert recognized schemas to the latest + internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources' + type: string + kind: + description: 'Kind is a string value representing the REST resource this + object represents. Servers may infer this from the endpoint the client + submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds' + type: string + metadata: + type: object + spec: + description: ModelSpec defines the desired state of Model + properties: + creator: + description: Creator defines dataset creator(AUTO-FILLED by webhook) + type: string + description: + description: Description defines datasource description + type: string + displayName: + description: DisplayName defines dataset display name + type: string + type: + description: Type defines what kind of model this is + type: string + required: + - displayName + type: object + status: + description: ModelStatus defines the observed state of Model + properties: + conditions: + description: Conditions of the resource. + items: + description: A Condition that may apply to a resource. + properties: + lastSuccessfulTime: + description: LastSuccessfulTime is repository Last Successful + Update Time + format: date-time + type: string + lastTransitionTime: + description: LastTransitionTime is the last time this condition + transitioned from one status to another. + format: date-time + type: string + message: + description: A Message containing details about this condition's + last transition from one status to another, if any. + type: string + reason: + description: A Reason for this condition's last transition from + one status to another. + type: string + status: + description: Status of this condition; is it currently True, + False, or Unknown + type: string + type: + description: Type of this condition. At most one of each condition + type may apply to a resource at any point in time. + type: string + required: + - lastTransitionTime + - reason + - status + - type + type: object + type: array + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_versioneddatasets.yaml b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_versioneddatasets.yaml index b2c83eb0d..9f06a7ea6 100644 --- a/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_versioneddatasets.yaml +++ b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_versioneddatasets.yaml @@ -64,11 +64,12 @@ spec: - kind - name type: object - files: - description: Files included in this `VersionedDataset` + fileGroups: + description: FileGroups included in this `VersionedDataset` Grouped + by Datasource items: properties: - from: + datasource: description: From defines the datasource which provides this `File` properties: @@ -92,12 +93,15 @@ spec: - kind - name type: object - path: - description: Path defines the detail path to get this `File` - type: string + paths: + description: Paths defines the detail paths to get objects from + above datasource + items: + type: string + type: array required: - - from - - path + - datasource + - paths type: object type: array version: @@ -148,84 +152,6 @@ spec: - type type: object type: array - filesStatus: - description: FilesStatus contains the status to all files in VersionedDatasetSpec - items: - properties: - processCondition: - description: ProcessCondition records the status of data processing - properties: - lastSuccessfulTime: - description: LastSuccessfulTime is repository Last Successful - Update Time - format: date-time - type: string - lastTransitionTime: - description: LastTransitionTime is the last time this condition - transitioned from one status to another. - format: date-time - type: string - message: - description: A Message containing details about this condition's - last transition from one status to another, if any. - type: string - reason: - description: A Reason for this condition's last transition - from one status to another. - type: string - status: - description: Status of this condition; is it currently True, - False, or Unknown - type: string - type: - description: Type of this condition. At most one of each - condition type may apply to a resource at any point in - time. - type: string - required: - - lastTransitionTime - - reason - - status - - type - type: object - uploadCondition: - description: UploadCondition records the status of file upload - properties: - lastSuccessfulTime: - description: LastSuccessfulTime is repository Last Successful - Update Time - format: date-time - type: string - lastTransitionTime: - description: LastTransitionTime is the last time this condition - transitioned from one status to another. - format: date-time - type: string - message: - description: A Message containing details about this condition's - last transition from one status to another, if any. - type: string - reason: - description: A Reason for this condition's last transition - from one status to another. - type: string - status: - description: Status of this condition; is it currently True, - False, or Unknown - type: string - type: - description: Type of this condition. At most one of each - condition type may apply to a resource at any point in - time. - type: string - required: - - lastTransitionTime - - reason - - status - - type - type: object - type: object - type: array type: object type: object served: true diff --git a/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_workers.yaml b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_workers.yaml new file mode 100644 index 000000000..fc1d1f4bf --- /dev/null +++ b/charts/arcadia/crds/arcadia.kubeagi.k8s.com.cn_workers.yaml @@ -0,0 +1,93 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.9.2 + creationTimestamp: null + name: workers.arcadia.kubeagi.k8s.com.cn +spec: + group: arcadia.kubeagi.k8s.com.cn + names: + kind: Worker + listKind: WorkerList + plural: workers + singular: worker + scope: Namespaced + versions: + - name: v1alpha1 + schema: + openAPIV3Schema: + description: Worker is the Schema for the workers API + properties: + apiVersion: + description: 'APIVersion defines the versioned schema of this representation + of an object. Servers should convert recognized schemas to the latest + internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources' + type: string + kind: + description: 'Kind is a string value representing the REST resource this + object represents. Servers may infer this from the endpoint the client + submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds' + type: string + metadata: + type: object + spec: + description: WorkerSpec defines the desired state of Worker + properties: + creator: + description: Creator defines dataset creator(AUTO-FILLED by webhook) + type: string + displayName: + description: DisplayName defines dataset display name + type: string + required: + - displayName + type: object + status: + description: WorkerStatus defines the observed state of Worker + properties: + conditions: + description: Conditions of the resource. + items: + description: A Condition that may apply to a resource. + properties: + lastSuccessfulTime: + description: LastSuccessfulTime is repository Last Successful + Update Time + format: date-time + type: string + lastTransitionTime: + description: LastTransitionTime is the last time this condition + transitioned from one status to another. + format: date-time + type: string + message: + description: A Message containing details about this condition's + last transition from one status to another, if any. + type: string + reason: + description: A Reason for this condition's last transition from + one status to another. + type: string + status: + description: Status of this condition; is it currently True, + False, or Unknown + type: string + type: + description: Type of this condition. At most one of each condition + type may apply to a resource at any point in time. + type: string + required: + - lastTransitionTime + - reason + - status + - type + type: object + type: array + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/charts/arcadia/templates/rbac.yaml b/charts/arcadia/templates/rbac.yaml index 8cf3b69a3..e15f708bf 100644 --- a/charts/arcadia/templates/rbac.yaml +++ b/charts/arcadia/templates/rbac.yaml @@ -168,6 +168,32 @@ rules: - get - patch - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models/finalizers + verbs: + - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models/status + verbs: + - get + - patch + - update - apiGroups: - arcadia.kubeagi.k8s.com.cn resources: @@ -220,3 +246,29 @@ rules: - get - patch - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers/finalizers + verbs: + - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers/status + verbs: + - get + - patch + - update diff --git a/config/crd/bases/arcadia.kubeagi.k8s.com.cn_datasources.yaml b/config/crd/bases/arcadia.kubeagi.k8s.com.cn_datasources.yaml index 590497c12..2883a043d 100644 --- a/config/crd/bases/arcadia.kubeagi.k8s.com.cn_datasources.yaml +++ b/config/crd/bases/arcadia.kubeagi.k8s.com.cn_datasources.yaml @@ -92,6 +92,7 @@ spec: bucket: type: string object: + description: Object must end with a slash "/" if it is a directory type: string type: object required: diff --git a/config/crd/bases/arcadia.kubeagi.k8s.com.cn_models.yaml b/config/crd/bases/arcadia.kubeagi.k8s.com.cn_models.yaml new file mode 100644 index 000000000..50434d738 --- /dev/null +++ b/config/crd/bases/arcadia.kubeagi.k8s.com.cn_models.yaml @@ -0,0 +1,99 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.9.2 + creationTimestamp: null + name: models.arcadia.kubeagi.k8s.com.cn +spec: + group: arcadia.kubeagi.k8s.com.cn + names: + kind: Model + listKind: ModelList + plural: models + singular: model + scope: Namespaced + versions: + - name: v1alpha1 + schema: + openAPIV3Schema: + description: Model is the Schema for the models API + properties: + apiVersion: + description: 'APIVersion defines the versioned schema of this representation + of an object. Servers should convert recognized schemas to the latest + internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources' + type: string + kind: + description: 'Kind is a string value representing the REST resource this + object represents. Servers may infer this from the endpoint the client + submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds' + type: string + metadata: + type: object + spec: + description: ModelSpec defines the desired state of Model + properties: + creator: + description: Creator defines dataset creator(AUTO-FILLED by webhook) + type: string + description: + description: Description defines datasource description + type: string + displayName: + description: DisplayName defines dataset display name + type: string + type: + description: Type defines what kind of model this is + type: string + required: + - displayName + type: object + status: + description: ModelStatus defines the observed state of Model + properties: + conditions: + description: Conditions of the resource. + items: + description: A Condition that may apply to a resource. + properties: + lastSuccessfulTime: + description: LastSuccessfulTime is repository Last Successful + Update Time + format: date-time + type: string + lastTransitionTime: + description: LastTransitionTime is the last time this condition + transitioned from one status to another. + format: date-time + type: string + message: + description: A Message containing details about this condition's + last transition from one status to another, if any. + type: string + reason: + description: A Reason for this condition's last transition from + one status to another. + type: string + status: + description: Status of this condition; is it currently True, + False, or Unknown + type: string + type: + description: Type of this condition. At most one of each condition + type may apply to a resource at any point in time. + type: string + required: + - lastTransitionTime + - reason + - status + - type + type: object + type: array + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/config/crd/bases/arcadia.kubeagi.k8s.com.cn_workers.yaml b/config/crd/bases/arcadia.kubeagi.k8s.com.cn_workers.yaml new file mode 100644 index 000000000..fc1d1f4bf --- /dev/null +++ b/config/crd/bases/arcadia.kubeagi.k8s.com.cn_workers.yaml @@ -0,0 +1,93 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.9.2 + creationTimestamp: null + name: workers.arcadia.kubeagi.k8s.com.cn +spec: + group: arcadia.kubeagi.k8s.com.cn + names: + kind: Worker + listKind: WorkerList + plural: workers + singular: worker + scope: Namespaced + versions: + - name: v1alpha1 + schema: + openAPIV3Schema: + description: Worker is the Schema for the workers API + properties: + apiVersion: + description: 'APIVersion defines the versioned schema of this representation + of an object. Servers should convert recognized schemas to the latest + internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources' + type: string + kind: + description: 'Kind is a string value representing the REST resource this + object represents. Servers may infer this from the endpoint the client + submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds' + type: string + metadata: + type: object + spec: + description: WorkerSpec defines the desired state of Worker + properties: + creator: + description: Creator defines dataset creator(AUTO-FILLED by webhook) + type: string + displayName: + description: DisplayName defines dataset display name + type: string + required: + - displayName + type: object + status: + description: WorkerStatus defines the observed state of Worker + properties: + conditions: + description: Conditions of the resource. + items: + description: A Condition that may apply to a resource. + properties: + lastSuccessfulTime: + description: LastSuccessfulTime is repository Last Successful + Update Time + format: date-time + type: string + lastTransitionTime: + description: LastTransitionTime is the last time this condition + transitioned from one status to another. + format: date-time + type: string + message: + description: A Message containing details about this condition's + last transition from one status to another, if any. + type: string + reason: + description: A Reason for this condition's last transition from + one status to another. + type: string + status: + description: Status of this condition; is it currently True, + False, or Unknown + type: string + type: + description: Type of this condition. At most one of each condition + type may apply to a resource at any point in time. + type: string + required: + - lastTransitionTime + - reason + - status + - type + type: object + type: array + type: object + type: object + served: true + storage: true + subresources: + status: {} diff --git a/config/crd/kustomization.yaml b/config/crd/kustomization.yaml index df49548ff..8af4c452b 100644 --- a/config/crd/kustomization.yaml +++ b/config/crd/kustomization.yaml @@ -9,6 +9,8 @@ resources: - bases/arcadia.kubeagi.k8s.com.cn_embedders.yaml - bases/arcadia.kubeagi.k8s.com.cn_datasets.yaml - bases/arcadia.kubeagi.k8s.com.cn_versioneddatasets.yaml +- bases/arcadia.kubeagi.k8s.com.cn_workers.yaml +- bases/arcadia.kubeagi.k8s.com.cn_models.yaml #+kubebuilder:scaffold:crdkustomizeresource patchesStrategicMerge: @@ -21,6 +23,8 @@ patchesStrategicMerge: #- patches/webhook_in_embedders.yaml #- patches/webhook_in_datasets.yaml #- patches/webhook_in_versioneddatasets.yaml +#- patches/webhook_in_workers.yaml +#- patches/webhook_in_models.yaml #+kubebuilder:scaffold:crdkustomizewebhookpatch # [CERTMANAGER] To enable cert-manager, uncomment all the sections with [CERTMANAGER] prefix. @@ -32,6 +36,8 @@ patchesStrategicMerge: #- patches/cainjection_in_embedders.yaml #- patches/cainjection_in_datasets.yaml #- patches/cainjection_in_versioneddatasets.yaml +#- patches/cainjection_in_workers.yaml +#- patches/cainjection_in_models.yaml #+kubebuilder:scaffold:crdkustomizecainjectionpatch # the following config is for teaching kustomize how to do kustomization for CRDs. diff --git a/config/crd/patches/cainjection_in_models.yaml b/config/crd/patches/cainjection_in_models.yaml new file mode 100644 index 000000000..e00b8fe8f --- /dev/null +++ b/config/crd/patches/cainjection_in_models.yaml @@ -0,0 +1,7 @@ +# The following patch adds a directive for certmanager to inject CA into the CRD +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + cert-manager.io/inject-ca-from: $(CERTIFICATE_NAMESPACE)/$(CERTIFICATE_NAME) + name: models.arcadia.kubeagi.k8s.com.cn diff --git a/config/crd/patches/cainjection_in_workers.yaml b/config/crd/patches/cainjection_in_workers.yaml new file mode 100644 index 000000000..246015070 --- /dev/null +++ b/config/crd/patches/cainjection_in_workers.yaml @@ -0,0 +1,7 @@ +# The following patch adds a directive for certmanager to inject CA into the CRD +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + cert-manager.io/inject-ca-from: $(CERTIFICATE_NAMESPACE)/$(CERTIFICATE_NAME) + name: workers.arcadia.kubeagi.k8s.com.cn diff --git a/config/crd/patches/webhook_in_models.yaml b/config/crd/patches/webhook_in_models.yaml new file mode 100644 index 000000000..629fd8322 --- /dev/null +++ b/config/crd/patches/webhook_in_models.yaml @@ -0,0 +1,16 @@ +# The following patch enables a conversion webhook for the CRD +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: models.arcadia.kubeagi.k8s.com.cn +spec: + conversion: + strategy: Webhook + webhook: + clientConfig: + service: + namespace: system + name: webhook-service + path: /convert + conversionReviewVersions: + - v1 diff --git a/config/crd/patches/webhook_in_workers.yaml b/config/crd/patches/webhook_in_workers.yaml new file mode 100644 index 000000000..fb55b1899 --- /dev/null +++ b/config/crd/patches/webhook_in_workers.yaml @@ -0,0 +1,16 @@ +# The following patch enables a conversion webhook for the CRD +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: workers.arcadia.kubeagi.k8s.com.cn +spec: + conversion: + strategy: Webhook + webhook: + clientConfig: + service: + namespace: system + name: webhook-service + path: /convert + conversionReviewVersions: + - v1 diff --git a/config/rbac/model_editor_role.yaml b/config/rbac/model_editor_role.yaml new file mode 100644 index 000000000..2cbf81d05 --- /dev/null +++ b/config/rbac/model_editor_role.yaml @@ -0,0 +1,24 @@ +# permissions for end users to edit models. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: model-editor-role +rules: +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models/status + verbs: + - get diff --git a/config/rbac/model_viewer_role.yaml b/config/rbac/model_viewer_role.yaml new file mode 100644 index 000000000..6e4237895 --- /dev/null +++ b/config/rbac/model_viewer_role.yaml @@ -0,0 +1,20 @@ +# permissions for end users to view models. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: model-viewer-role +rules: +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models + verbs: + - get + - list + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models/status + verbs: + - get diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 16536a455..7d6d40f26 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -151,6 +151,32 @@ rules: - get - patch - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models/finalizers + verbs: + - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - models/status + verbs: + - get + - patch + - update - apiGroups: - arcadia.kubeagi.k8s.com.cn resources: @@ -203,3 +229,29 @@ rules: - get - patch - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers/finalizers + verbs: + - update +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers/status + verbs: + - get + - patch + - update diff --git a/config/rbac/worker_editor_role.yaml b/config/rbac/worker_editor_role.yaml new file mode 100644 index 000000000..783dda724 --- /dev/null +++ b/config/rbac/worker_editor_role.yaml @@ -0,0 +1,24 @@ +# permissions for end users to edit workers. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: worker-editor-role +rules: +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers + verbs: + - create + - delete + - get + - list + - patch + - update + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers/status + verbs: + - get diff --git a/config/rbac/worker_viewer_role.yaml b/config/rbac/worker_viewer_role.yaml new file mode 100644 index 000000000..b9009c0bd --- /dev/null +++ b/config/rbac/worker_viewer_role.yaml @@ -0,0 +1,20 @@ +# permissions for end users to view workers. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: worker-viewer-role +rules: +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers + verbs: + - get + - list + - watch +- apiGroups: + - arcadia.kubeagi.k8s.com.cn + resources: + - workers/status + verbs: + - get diff --git a/config/samples/arcadia_v1alpha1_local_datasource.yaml b/config/samples/arcadia_v1alpha1_local_datasource.yaml index 0e7005ea2..89d409d76 100644 --- a/config/samples/arcadia_v1alpha1_local_datasource.yaml +++ b/config/samples/arcadia_v1alpha1_local_datasource.yaml @@ -5,4 +5,4 @@ metadata: namespace: arcadia spec: displayName: "本地存储数据源" - description: "这是一个使用系统数据源作为存储的本地数据源" \ No newline at end of file + description: "这是一个由本地上传的文件组成的数据源" \ No newline at end of file diff --git a/config/samples/arcadia_v1alpha1_model_chatglm2-6b.yaml b/config/samples/arcadia_v1alpha1_model_chatglm2-6b.yaml new file mode 100644 index 000000000..016b738d9 --- /dev/null +++ b/config/samples/arcadia_v1alpha1_model_chatglm2-6b.yaml @@ -0,0 +1,9 @@ +apiVersion: arcadia.kubeagi.k8s.com.cn/v1alpha1 +kind: Model +metadata: + name: chatglm2-6b # name will be used as the obejct + namespace: arcadia +spec: + displayName: "chatglm2-6b" + description: "清华发布的ChatGLM2-6B模型" + type: "llm" diff --git a/config/samples/arcadia_v1alpha1_worker.yaml b/config/samples/arcadia_v1alpha1_worker.yaml new file mode 100644 index 000000000..6343e25d4 --- /dev/null +++ b/config/samples/arcadia_v1alpha1_worker.yaml @@ -0,0 +1,6 @@ +apiVersion: arcadia.kubeagi.k8s.com.cn/v1alpha1 +kind: Worker +metadata: + name: worker-sample +spec: + # TODO(user): Add fields here diff --git a/config/samples/kustomization.yaml b/config/samples/kustomization.yaml index eb170c0d6..b096ce099 100644 --- a/config/samples/kustomization.yaml +++ b/config/samples/kustomization.yaml @@ -7,4 +7,6 @@ resources: - arcadia_v1alpha1_embedders.yaml - arcadia_v1alpha1_dataset.yaml - arcadia_v1alpha1_versioneddataset.yaml +- arcadia_v1alpha1_worker.yaml +- arcadia_v1alpha1_model.yaml #+kubebuilder:scaffold:manifestskustomizesamples diff --git a/controllers/datasource_controller.go b/controllers/datasource_controller.go index e98908dcd..2de8cfa6d 100644 --- a/controllers/datasource_controller.go +++ b/controllers/datasource_controller.go @@ -88,7 +88,7 @@ func (r *DatasourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) // initialize labels requeue, err := r.Initialize(ctx, logger, instance) if err != nil { - return reconcile.Result{}, fmt.Errorf("failed to initiali datasource: %w", err) + return reconcile.Result{}, fmt.Errorf("failed to initialize datasource: %w", err) } if requeue { return reconcile.Result{Requeue: true}, nil @@ -109,7 +109,8 @@ func (r *DatasourceReconciler) SetupWithManager(mgr ctrl.Manager) error { UpdateFunc: func(ue event.UpdateEvent) bool { oldDatsource := ue.ObjectOld.(*arcadiav1alpha1.Datasource) newDatasource := ue.ObjectNew.(*arcadiav1alpha1.Datasource) - return !reflect.DeepEqual(oldDatsource.Spec, newDatasource.Spec) + return !reflect.DeepEqual(oldDatsource.Spec, newDatasource.Spec) || + newDatasource.DeletionTimestamp != nil }, })). Complete(r) @@ -186,7 +187,7 @@ func (r *DatasourceReconciler) Checkdatasource(ctx context.Context, logger logr. } // check datasource - if err := ds.Check(ctx, info); err != nil { + if err := ds.Stat(ctx, info); err != nil { return r.UpdateStatus(ctx, instance, err) } diff --git a/controllers/model_controller.go b/controllers/model_controller.go new file mode 100644 index 000000000..185814a21 --- /dev/null +++ b/controllers/model_controller.go @@ -0,0 +1,258 @@ +/* +Copyright 2023 KubeAGI. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +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" + "github.com/kubeagi/arcadia/pkg/config" + "github.com/kubeagi/arcadia/pkg/datasource" + "github.com/kubeagi/arcadia/pkg/utils" +) + +// ModelReconciler reconciles a Model object +type ModelReconciler struct { + client.Client + Scheme *runtime.Scheme +} + +//+kubebuilder:rbac:groups=arcadia.kubeagi.k8s.com.cn,resources=models,verbs=get;list;watch;create;update;patch;delete +//+kubebuilder:rbac:groups=arcadia.kubeagi.k8s.com.cn,resources=models/status,verbs=get;update;patch +//+kubebuilder:rbac:groups=arcadia.kubeagi.k8s.com.cn,resources=models/finalizers,verbs=update +//+kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch +//+kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch + +// Reconcile is part of the main kubernetes reconciliation loop which aims to +// move the current state of the cluster closer to the desired state. +// TODO(user): Modify the Reconcile function to compare the state specified by +// the Model object against the actual cluster state, and then +// perform operations to make the cluster state reflect the state specified by +// the user. +// +// For more details, check Reconcile and its Result here: +// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.12.2/pkg/reconcile +func (r *ModelReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + _ = log.FromContext(ctx) + + logger := log.FromContext(ctx) + logger.Info("Starting model reconcile") + + instance := &arcadiav1alpha1.Model{} + if err := r.Get(ctx, req.NamespacedName, instance); err != nil { + if errors.IsNotFound(err) { + // model has been deleted. + return reconcile.Result{}, nil + } + return reconcile.Result{}, err + } + + if instance.DeletionTimestamp != nil { + logger.Info("Delete model") + // remove all model files from storage service + if err := r.RemoveModel(ctx, logger, instance); err != nil { + return reconcile.Result{}, fmt.Errorf("failed to remove model: %w", err) + } + // remove the finalizer to complete the delete action + instance.Finalizers = utils.RemoveString(instance.Finalizers, arcadiav1alpha1.Finalizer) + err := r.Client.Update(ctx, instance) + if err != nil { + return reconcile.Result{}, fmt.Errorf("failed to update model finializer: %w", err) + } + return reconcile.Result{}, nil + } + + // initialize labels + requeue, err := r.Initialize(ctx, logger, instance) + if err != nil { + return reconcile.Result{}, fmt.Errorf("failed to initialize model: %w", err) + } + if requeue { + return reconcile.Result{Requeue: true}, nil + } + + if err := r.CheckModel(ctx, logger, instance); err != nil { + // Update conditioned status + return reconcile.Result{}, err + } + + return ctrl.Result{}, nil +} + +// SetupWithManager sets up the controller with the Manager. +func (r *ModelReconciler) SetupWithManager(mgr ctrl.Manager) error { + return ctrl.NewControllerManagedBy(mgr). + For(&arcadiav1alpha1.Model{}, builder.WithPredicates(predicate.Funcs{ + UpdateFunc: func(ue event.UpdateEvent) bool { + oldModel := ue.ObjectOld.(*arcadiav1alpha1.Model) + newModel := ue.ObjectNew.(*arcadiav1alpha1.Model) + return !reflect.DeepEqual(oldModel.Spec, newModel.Spec) || newModel.DeletionTimestamp != nil + }, + })). + Complete(r) +} + +func (r *ModelReconciler) Initialize(ctx context.Context, logger logr.Logger, instance *arcadiav1alpha1.Model) (bool, error) { + instanceDeepCopy := instance.DeepCopy() + l := len(instanceDeepCopy.Finalizers) + + var update bool + + instanceDeepCopy.Finalizers = utils.AddString(instanceDeepCopy.Finalizers, arcadiav1alpha1.Finalizer) + if l != len(instanceDeepCopy.Finalizers) { + logger.V(1).Info("Add Finalizer for model", "Finalizer", arcadiav1alpha1.Finalizer) + update = true + } + + // Initialize Labels + if instanceDeepCopy.Labels == nil { + instanceDeepCopy.Labels = make(map[string]string) + } + // For model type + currentType := string(instanceDeepCopy.ModelType()) + if v := instanceDeepCopy.Labels[arcadiav1alpha1.LabelModelType]; v != currentType { + instanceDeepCopy.Labels[arcadiav1alpha1.LabelModelType] = currentType + update = true + } + + // Initialize annotations + if instanceDeepCopy.Annotations == nil { + instanceDeepCopy.Annotations = make(map[string]string) + } + // For model's full storeage path + currentFullPath := instanceDeepCopy.FullPath() + if v := instanceDeepCopy.Annotations[arcadiav1alpha1.LabelModelFullPath]; v != currentFullPath { + instanceDeepCopy.Annotations[arcadiav1alpha1.LabelModelFullPath] = currentFullPath + update = true + } + + if update { + return true, r.Client.Update(ctx, instanceDeepCopy) + } + + return false, nil +} + +// CheckModel to update status +func (r *ModelReconciler) CheckModel(ctx context.Context, logger logr.Logger, instance *arcadiav1alpha1.Model) error { + logger.Info("check model") + var err error + + var ds datasource.Datasource + var info any + + system, err := config.GetSystemDatasource(ctx, r.Client) + if err != nil { + return r.UpdateStatus(ctx, instance, err) + } + endpoint := system.Spec.Enpoint.DeepCopy() + if endpoint != nil && endpoint.AuthSecret != nil { + endpoint.AuthSecret.WithNameSpace(system.Namespace) + } + ds, err = datasource.NewLocal(ctx, r.Client, endpoint) + if err != nil { + return r.UpdateStatus(ctx, instance, err) + } + // oss info: + // - bucket: same as the instance namespace + // - object: path joined with "model/{instance.name}" + info = &arcadiav1alpha1.OSS{ + Bucket: instance.Namespace, + Object: instance.ObjectPath(), + } + + // check datasource against info + if err := ds.Stat(ctx, info); err != nil { + return r.UpdateStatus(ctx, instance, err) + } + + // update status + return r.UpdateStatus(ctx, instance, nil) +} + +// Remove model files from storage +func (r *ModelReconciler) RemoveModel(ctx context.Context, logger logr.Logger, instance *arcadiav1alpha1.Model) error { + var ds datasource.Datasource + var info any + + system, err := config.GetSystemDatasource(ctx, r.Client) + if err != nil { + return r.UpdateStatus(ctx, instance, err) + } + endpoint := system.Spec.Enpoint.DeepCopy() + if endpoint != nil && endpoint.AuthSecret != nil { + endpoint.AuthSecret.WithNameSpace(system.Namespace) + } + ds, err = datasource.NewLocal(ctx, r.Client, endpoint) + if err != nil { + return r.UpdateStatus(ctx, instance, err) + } + + info = &arcadiav1alpha1.OSS{ + Bucket: instance.Namespace, + Object: instance.ObjectPath(), + } + + if err := ds.Stat(ctx, info); err != nil { + return nil + } + + return ds.Remove(ctx, info) +} + +// UpdateStatus uppon error +func (r *ModelReconciler) UpdateStatus(ctx context.Context, instance *arcadiav1alpha1.Model, err error) error { + instanceCopy := instance.DeepCopy() + var newCondition arcadiav1alpha1.Condition + if err != nil { + // set condition to False + newCondition = arcadiav1alpha1.Condition{ + Type: arcadiav1alpha1.TypeReady, + Status: corev1.ConditionFalse, + Reason: arcadiav1alpha1.ReasonUnavailable, + Message: err.Error(), + LastTransitionTime: metav1.Now(), + } + } else { + // set condition to True + newCondition = arcadiav1alpha1.Condition{ + Type: arcadiav1alpha1.TypeReady, + Status: corev1.ConditionTrue, + Reason: arcadiav1alpha1.ReasonAvailable, + Message: "Check Success", + LastTransitionTime: metav1.Now(), + LastSuccessfulTime: metav1.Now(), + } + } + instanceCopy.Status.SetConditions(newCondition) + return r.Client.Status().Update(ctx, instanceCopy) +} diff --git a/controllers/worker_controller.go b/controllers/worker_controller.go new file mode 100644 index 000000000..bd7523805 --- /dev/null +++ b/controllers/worker_controller.go @@ -0,0 +1,62 @@ +/* +Copyright 2023 KubeAGI. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controllers + +import ( + "context" + + "k8s.io/apimachinery/pkg/runtime" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" + + arcadiav1alpha1 "github.com/kubeagi/arcadia/api/v1alpha1" +) + +// WorkerReconciler reconciles a Worker object +type WorkerReconciler struct { + client.Client + Scheme *runtime.Scheme +} + +//+kubebuilder:rbac:groups=arcadia.kubeagi.k8s.com.cn,resources=workers,verbs=get;list;watch;create;update;patch;delete +//+kubebuilder:rbac:groups=arcadia.kubeagi.k8s.com.cn,resources=workers/status,verbs=get;update;patch +//+kubebuilder:rbac:groups=arcadia.kubeagi.k8s.com.cn,resources=workers/finalizers,verbs=update + +// Reconcile is part of the main kubernetes reconciliation loop which aims to +// move the current state of the cluster closer to the desired state. +// TODO(user): Modify the Reconcile function to compare the state specified by +// the Worker object against the actual cluster state, and then +// perform operations to make the cluster state reflect the state specified by +// the user. +// +// For more details, check Reconcile and its Result here: +// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.12.2/pkg/reconcile +func (r *WorkerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + _ = log.FromContext(ctx) + + // TODO(user): your logic here + + return ctrl.Result{}, nil +} + +// SetupWithManager sets up the controller with the Manager. +func (r *WorkerReconciler) SetupWithManager(mgr ctrl.Manager) error { + return ctrl.NewControllerManagedBy(mgr). + For(&arcadiav1alpha1.Worker{}). + Complete(r) +} diff --git a/main.go b/main.go index 5655a6bfa..e8de0c0fd 100644 --- a/main.go +++ b/main.go @@ -173,6 +173,20 @@ func main() { setupLog.Error(err, "unable to create controller", "controller", "VersionedDataset") os.Exit(1) } + if err = (&controllers.WorkerReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "Worker") + os.Exit(1) + } + if err = (&controllers.ModelReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "Model") + os.Exit(1) + } //+kubebuilder:scaffold:builder if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { diff --git a/pkg/datasource/datasource.go b/pkg/datasource/datasource.go index b61f8002c..8b4a2c4a2 100644 --- a/pkg/datasource/datasource.go +++ b/pkg/datasource/datasource.go @@ -19,6 +19,7 @@ package datasource import ( "context" "errors" + "strings" "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" @@ -31,11 +32,14 @@ import ( var ( ErrUnknowDatasourceType = errors.New("unknow datasource type") - ErrOSSNoSuchBucket = errors.New("NoSuchBucket") + ErrBucketNotProvided = errors.New("no bucket provided") + ErrOSSNoSuchBucket = errors.New("no such bucket") + ErrOSSNoSuchObject = errors.New("no such object in bucket") ) type Datasource interface { - Check(ctx context.Context, info any) error + Stat(ctx context.Context, info any) error + Remove(ctx context.Context, info any) error } var _ Datasource = (*Unknown)(nil) @@ -47,7 +51,11 @@ func NewUnknown(ctx context.Context, c client.Client) (*Unknown, error) { return &Unknown{}, nil } -func (u *Unknown) Check(ctx context.Context, info any) error { +func (u *Unknown) Stat(ctx context.Context, info any) error { + return ErrUnknowDatasourceType +} + +func (u *Unknown) Remove(ctx context.Context, info any) error { return ErrUnknowDatasourceType } @@ -67,13 +75,13 @@ func NewLocal(ctx context.Context, c client.Client, endpoint *v1alpha1.Endpoint) return &Local{oss: oss}, nil } -// Check `Local` with `OSS` -func (local *Local) Check(ctx context.Context, options any) (err error) { - err = local.oss.Check(ctx, options) +// Stat `Local` with `OSS` +func (local *Local) Stat(ctx context.Context, options any) (err error) { + err = local.oss.Stat(ctx, options) if err != nil && errors.Is(err, ErrOSSNoSuchBucket) { ossInfo, ok := options.(*v1alpha1.OSS) if !ok { - return errors.New("invalid check info for OSS") + return errors.New("invalid stat info for OSS") } defautlMakeBucketOptions := minio.MakeBucketOptions{} err = local.oss.MakeBucket(ctx, ossInfo.Bucket, defautlMakeBucketOptions) @@ -81,6 +89,11 @@ func (local *Local) Check(ctx context.Context, options any) (err error) { return err } +// Remove object from OSS +func (local *Local) Remove(ctx context.Context, info any) error { + return local.oss.Remove(ctx, info) +} + var _ Datasource = (*OSS)(nil) // OSS is a wrapper to object storage service @@ -118,7 +131,7 @@ func NewOSS(ctx context.Context, c client.Client, endpoint *v1alpha1.Endpoint) ( } // Check oss agains info() -func (oss *OSS) Check(ctx context.Context, info any) error { +func (oss *OSS) Stat(ctx context.Context, info any) error { if info == nil { return nil } @@ -127,21 +140,49 @@ func (oss *OSS) Check(ctx context.Context, info any) error { return errors.New("invalid check info for OSS") } - if ossInfo.Bucket != "" { - exist, err := oss.Client.BucketExists(ctx, ossInfo.Bucket) - if err != nil { - return err - } - if !exist { - return ErrOSSNoSuchBucket - } + return oss.statObject(ctx, ossInfo) +} + +// TODO: implement `Remove` against info +func (oss *OSS) Remove(ctx context.Context, info any) error { + return nil +} - if ossInfo.Object != "" { - _, err := oss.Client.StatObject(ctx, ossInfo.Bucket, ossInfo.Object, minio.StatObjectOptions{}) - if err != nil { - return err +// StatObject against oss info +// Q: Why not using client.StatObject() ? +// A: The `StateObject()` won't treat path(directory) as a valid object +func (oss *OSS) statObject(ctx context.Context, ossInfo *v1alpha1.OSS) error { + if ossInfo.Bucket == "" { + return ErrBucketNotProvided + } + + // check whether bucket exists + isExist, err := oss.Client.BucketExists(ctx, ossInfo.Bucket) + if err != nil { + return err + } + if !isExist { + return ErrOSSNoSuchBucket + } + + // check whether object exists + if ossInfo.Object != "" { + // The object by `ListObjects` will trim "/" automatically,so we also need to trim "/" to make sure name comparision successful + ossInfo.Object = strings.TrimPrefix(ossInfo.Object, "/") + // When object contains "/" which means it is a directory,'ListObjects' will show all objects under that directory without object itself + // After we remove "/", the objects by `ListObjects` will have object itself included. + trimmedObjectPath := strings.TrimSuffix(ossInfo.Object, "/") + for objInfo := range oss.Client.ListObjects( + ctx, ossInfo.Bucket, minio.ListObjectsOptions{ + Prefix: trimmedObjectPath, + }, + ) { + if objInfo.Key == ossInfo.Object { + return nil } } + return ErrOSSNoSuchObject } + return nil }