Commit 5ddc7f7d authored by Maksim Khadkevich's avatar Maksim Khadkevich Committed by GitHub
Browse files

feat: moved compoundAI operator, APIserver, and examples (#10)

parent 14ce7e03
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type DeploymentStatus string
const (
DeploymentStatusUnknown DeploymentStatus = "unknown"
DeploymentStatusNonDeployed DeploymentStatus = "non-deployed"
DeploymentStatusRunning DeploymentStatus = "running"
DeploymentStatusUnhealthy DeploymentStatus = "unhealthy"
DeploymentStatusFailed DeploymentStatus = "failed"
DeploymentStatusDeploying DeploymentStatus = "deploying"
DeploymentStatusTerminating DeploymentStatus = "terminating"
DeploymentStatusTerminated DeploymentStatus = "terminated"
DeploymentStatusImageBuilding DeploymentStatus = "image-building"
DeploymentStatusImageBuildFailed DeploymentStatus = "image-build-failed"
DeploymentStatusImageBuildSucceeded DeploymentStatus = "image-build-succeeded"
)
func (d DeploymentStatus) Ptr() *DeploymentStatus {
return &d
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
import (
"database/sql/driver"
"encoding/json"
"fmt"
)
type DeploymentTargetType string
const (
DeploymentTargetTypeStable DeploymentTargetType = "stable"
DeploymentTargetTypeCanary DeploymentTargetType = "canary"
)
type DeploymentStrategy string
const (
DeploymentStrategyRollingUpdate DeploymentStrategy = "RollingUpdate"
DeploymentStrategyRecreate DeploymentStrategy = "Recreate"
DeploymentStrategyRampedSlowRollout DeploymentStrategy = "RampedSlowRollout"
DeploymentStrategyBestEffortControlledRollout DeploymentStrategy = "BestEffortControlledRollout"
)
var DeploymentTargetTypeAddrs = map[DeploymentTargetType]string{
DeploymentTargetTypeStable: "stb",
DeploymentTargetTypeCanary: "cnr",
}
type DeploymentTargetHPAConf struct {
CPU *int32 `json:"cpu,omitempty"`
GPU *int32 `json:"gpu,omitempty"`
Memory *string `json:"memory,omitempty"`
QPS *int64 `json:"qps,omitempty"`
MinReplicas *int32 `json:"min_replicas,omitempty"`
MaxReplicas *int32 `json:"max_replicas,omitempty"`
}
type DeploymentOverrides struct {
ColdStartTimeout *int32 `json:"cold_start_timeout"`
}
type DeploymentTargetConfig struct {
KubeResourceUid string `json:"kubeResourceUid"`
KubeResourceVersion string `json:"kubeResourceVersion"`
Resources *Resources `json:"resources"`
HPAConf *DeploymentTargetHPAConf `json:"hpa_conf,omitempty"`
EnableIngress *bool `json:"enable_ingress,omitempty"`
EnableStealingTrafficDebugMode *bool `json:"enable_stealing_traffic_debug_mode,omitempty"`
EnableDebugMode *bool `json:"enable_debug_mode,omitempty"`
EnableDebugPodReceiveProductionTraffic *bool `json:"enable_debug_pod_receive_production_traffic,omitempty"`
DeploymentStrategy *DeploymentStrategy `json:"deployment_strategy,omitempty"`
ExternalServices map[string]ExternalService `json:"external_services,omitempty"`
DeploymentOverrides *DeploymentOverrides `json:"DeploymentOverrides,omitempty"`
}
type CreateDeploymentTargetSchema struct {
CompoundNim string `json:"bento_repository"`
Version string `json:"bento"`
Config *DeploymentTargetConfig `json:"config"`
}
func (c *DeploymentTargetConfig) Scan(value interface{}) error {
if value == nil {
return nil
}
var data []byte
switch v := value.(type) {
case string:
data = []byte(v)
case []byte:
data = v
default:
return fmt.Errorf("unsupported type: %T", value)
}
return json.Unmarshal(data, c)
}
func (c *DeploymentTargetConfig) Value() (driver.Value, error) {
if c == nil {
return nil, nil
}
return json.Marshal(c)
}
type DeploymentTargetTypeSchema struct {
Type string `json:"type" enum:"stable,canary"`
}
type DeploymentTargetSchema struct {
ResourceSchema
DeploymentTargetTypeSchema
Creator *UserSchema `json:"creator"`
CompoundNimVersion *CompoundNimVersionFullSchema `json:"bento"`
Config *DeploymentTargetConfig `json:"config"`
}
type DeploymentTargetListSchema struct {
BaseListSchema
Items []*DeploymentTargetSchema `json:"items"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
import (
"time"
)
type EventStatus string
const (
EventStatusPending EventStatus = "pending"
EventStatusSuccess EventStatus = "success"
EventStatusFailed EventStatus = "failed"
)
type EventSchema struct {
BaseSchema
Resource interface{} `json:"resource,omitempty"`
Name string `json:"name,omitempty"`
Status EventStatus `json:"status,omitempty"`
OperationName string `json:"operation_name,omitempty"`
ApiTokenName string `json:"api_token_name,omitempty"`
Creator *UserSchema `json:"creator,omitempty"`
CreatedAt time.Time `json:"created_at,omitempty"`
ResourceDeleted bool `json:"resource_deleted,omitempty"`
}
type EventListSchema struct {
BaseListSchema
Items []*EventSchema `json:"items"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
import (
"encoding/json"
)
type ExternalService struct {
DeploymentSelectorKey string `json:"-"`
DeploymentSelectorValue string `json:"-"`
}
// UnmarshalJSON handles snake_case to struct mapping
func (e *ExternalService) UnmarshalJSON(data []byte) error {
var temp map[string]interface{}
if err := json.Unmarshal(data, &temp); err != nil {
return err
}
if val, ok := temp["deployment_selector_key"].(string); ok {
e.DeploymentSelectorKey = val
}
if val, ok := temp["deployment_selector_value"].(string); ok {
e.DeploymentSelectorValue = val
}
return nil
}
// MarshalJSON converts the struct to camelCase
func (e ExternalService) MarshalJSON() ([]byte, error) {
temp := map[string]interface{}{
"deploymentSelectorKey": e.DeploymentSelectorKey,
"deploymentSelectorValue": e.DeploymentSelectorValue,
}
return json.Marshal(temp)
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type LabelItemSchema struct {
Key string `json:"key"`
Value string `json:"value"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type BaseListSchema struct {
Total uint `json:"total"`
Start uint `json:"start"`
Count uint `json:"count"`
}
type ListQuerySchema struct {
Start uint `form:"start"`
Count uint `form:"count"`
Search *string `form:"search"`
Q string `query:"q"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type OrganizationSchema struct {
ResourceSchema
Creator *UserSchema `json:"creator"`
Description string `json:"description"`
}
type OrganizationFullSchema struct {
OrganizationSchema
}
type OrganizationListSchema struct {
BaseListSchema
Items []*OrganizationSchema `json:"items"`
}
type UpdateOrganizationSchema struct {
Description *string `json:"description"`
}
type CreateOrganizationSchema struct {
Name string `json:"name"`
Description string `json:"description"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type MemberRole string
const (
MemberRoleGuest MemberRole = "guest"
MemberRoleDeveloper MemberRole = "developer"
MemberRoleAdmin MemberRole = "admin"
)
type OrganizationMemberSchema struct {
BaseSchema
Role MemberRole `json:"role"`
Creator *UserSchema `json:"creator"`
User UserSchema `json:"user"`
Organization OrganizationSchema `json:"organization"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type OwnershipSchema struct {
OrganizationId string
UserId string
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type IResourceSchema interface {
GetType() ResourceType
GetName() string
}
type ResourceSchema struct {
BaseSchema
Name string `json:"name"`
Labels []LabelItemSchema `json:"labels"`
ResourceType ResourceType `json:"resource_type" enum:"user,organization,cluster,compound_nim,compound_nim_version,deployment,deployment_revision,model_repository,model,api_token"`
}
func (r ResourceSchema) GetType() ResourceType {
return r.ResourceType
}
func (r ResourceSchema) GetName() string {
return r.Name
}
func (s *ResourceSchema) TypeName() string {
return string(s.ResourceType)
}
type ResourceItem struct {
CPU string `json:"cpu,omitempty"`
Memory string `json:"memory,omitempty"`
GPU string `json:"gpu,omitempty"`
Custom map[string]string `json:"custom,omitempty"`
}
type Resources struct {
Requests *ResourceItem `json:"requests,omitempty"`
Limits *ResourceItem `json:"limits,omitempty"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type ResourceType string
const (
ResourceTypeUser ResourceType = "user"
ResourceTypeOrganization ResourceType = "organization"
ResourceTypeCluster ResourceType = "cluster"
ResourceTypeCompoundNim ResourceType = "compound_nim"
ResourceTypeCompoundNimVersion ResourceType = "compound_nim_version"
ResourceTypeDeployment ResourceType = "deployment"
ResourceTypeDeploymentRevision ResourceType = "deployment_revision"
ResourceTypeTerminalRecord ResourceType = "terminal_record"
ResourceTypeModelRepository ResourceType = "model_repository"
ResourceTypeModel ResourceType = "model"
ResourceTypeLabel ResourceType = "label"
ResourceTypeApiToken ResourceType = "api_token"
ResourceTypeCompoundAIComponent ResourceType = "yatai_component"
)
func (type_ ResourceType) Ptr() *ResourceType {
return &type_
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type UserSchema struct {
ResourceSchema
FirstName string `json:"first_name"`
LastName string `json:"last_name"`
Email string `json:"email"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemas
type VersionSchema struct {
Version string `json:"version"`
GitCommit string `json:"git_commit"`
BuildDate string `json:"build_date"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemasv2
import "github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/schemas"
type ClusterSchema struct {
schemas.ResourceSchema
Description string `json:"description"`
OrganizationName string `json:"organization_name"`
Creator *schemas.UserSchema `json:"creator"`
IsFirst *bool `json:"is_first,omitempty"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 schemasv2
import "github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/schemas"
type DeploymentSchema struct {
schemas.ResourceSchema
Creator *schemas.UserSchema `json:"creator"`
Cluster *ClusterSchema `json:"cluster"`
Status schemas.DeploymentStatus `json:"status" enum:"unknown,non-deployed,running,unhealthy,failed,deploying"`
URLs []string `json:"urls"`
LatestRevision *schemas.DeploymentRevisionSchema `json:"latest_revision"`
KubeNamespace string `json:"kube_namespace"`
}
type GetDeploymentSchema struct {
DeploymentName string `uri:"deploymentName" binding:"required"`
}
func (s *GetDeploymentSchema) ToV1(clusterName string, namespace string) *schemas.GetDeploymentSchema {
return &schemas.GetDeploymentSchema{
GetClusterSchema: schemas.GetClusterSchema{
ClusterName: clusterName,
},
KubeNamespace: namespace,
DeploymentName: s.DeploymentName,
}
}
type CreateDeploymentSchema struct {
UpdateDeploymentSchema
Name string `json:"name"`
}
type UpdateDeploymentSchema struct {
DeploymentConfigSchema
CompoundNim string `json:"bento"`
}
type DeploymentConfigSchema struct {
AccessAuthorization bool `json:"access_authorization"`
Envs interface{} `json:"envs,omitempty"`
Secrets interface{} `json:"secrets,omitempty"`
Services map[string]ServiceSpec `json:"services"`
}
type ServiceSpec struct {
Scaling ScalingSpec `json:"scaling"`
ConfigOverrides ConfigOverridesSpec `json:"config_overrides"`
ExternalServices map[string]schemas.ExternalService `json:"external_services,omitempty"`
ColdStartTimeout *int32 `json:"cold_start_timeout,omitempty"`
}
type ScalingSpec struct {
MinReplicas int `json:"min_replicas"`
MaxReplicas int `json:"max_replicas"`
}
type ConfigOverridesSpec struct {
Resources schemas.Resources `json:"resources"`
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 services
import (
"fmt"
"strings"
"gorm.io/gorm"
)
type BaseListOption struct {
Start *uint
Count *uint
Search *string
Keywords *[]string
KeywordFieldNames *[]string
}
func (opt BaseListOption) BindQueryWithLimit(query *gorm.DB) *gorm.DB {
if opt.Count != nil {
query = query.Limit(int(*opt.Count))
}
if opt.Start != nil {
query = query.Offset(int(*opt.Start))
}
return query
}
func (opt BaseListOption) BindQueryWithKeywords(query *gorm.DB, tableName string) *gorm.DB {
tableName = query.Statement.Quote(tableName)
keywordFieldNames := []string{"name"}
if opt.KeywordFieldNames != nil {
keywordFieldNames = *opt.KeywordFieldNames
}
if opt.Search != nil && *opt.Search != "" {
sqlPieces := make([]string, 0, len(keywordFieldNames))
args := make([]interface{}, 0, len(keywordFieldNames))
for _, keywordFieldName := range keywordFieldNames {
keywordFieldName = query.Statement.Quote(keywordFieldName)
sqlPieces = append(sqlPieces, fmt.Sprintf("%s.%s LIKE ?", tableName, keywordFieldName))
args = append(args, fmt.Sprintf("%%%s%%", *opt.Search))
}
query = query.Where(fmt.Sprintf("(%s)", strings.Join(sqlPieces, " OR ")), args...)
}
if opt.Keywords != nil {
sqlPieces := make([]string, 0, len(keywordFieldNames))
args := make([]interface{}, 0, len(keywordFieldNames))
for _, keywordFieldName := range keywordFieldNames {
keywordFieldName = query.Statement.Quote(keywordFieldName)
sqlPieces_ := make([]string, 0, len(*opt.Keywords))
for _, keyword := range *opt.Keywords {
sqlPieces_ = append(sqlPieces_, fmt.Sprintf("%s.%s LIKE ?", tableName, keywordFieldName))
args = append(args, fmt.Sprintf("%%%s%%", keyword))
}
sqlPieces = append(sqlPieces, fmt.Sprintf("(%s)", strings.Join(sqlPieces_, " AND ")))
}
query = query.Where(fmt.Sprintf("(%s)", strings.Join(sqlPieces, " OR ")), args...)
}
return query
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 services
import (
"context"
"errors"
"strings"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/common/consts"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/database"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/models"
"k8s.io/apimachinery/pkg/util/validation"
"github.com/rs/zerolog/log"
"gorm.io/gorm"
)
type clusterService struct{}
var ClusterService = clusterService{}
type CreateClusterOption struct {
CreatorId string
OrganizationId string
Name string
Description string
KubeConfig string
}
type UpdateClusterOption struct {
Description *string
KubeConfig *string
}
type ListClusterOption struct {
BaseListOption
OrganizationId *string
Ids *[]uint
Names *[]string
CreatorIds *[]uint
Order *string
}
func (s *clusterService) Create(ctx context.Context, opt CreateClusterOption) (*models.Cluster, error) {
errs := validation.IsDNS1035Label(opt.Name)
if len(errs) > 0 {
return nil, errors.New(strings.Join(errs, ";"))
}
db := s.getDB(ctx)
log.Info().Msg("Starting create cluster transaction")
cluster := models.Cluster{
Resource: models.Resource{
Name: opt.Name,
},
OrganizationAssociate: models.OrganizationAssociate{
OrganizationId: opt.OrganizationId,
},
CreatorAssociate: models.CreatorAssociate{
UserId: opt.CreatorId,
},
Description: opt.Description,
KubeConfig: opt.KubeConfig,
}
if err := db.Create(&cluster).Error; err != nil {
return nil, err
}
log.Info().Msg("Finished create cluster transaction")
return &cluster, nil
}
func (s *clusterService) Update(ctx context.Context, c *models.Cluster, opt UpdateClusterOption) (*models.Cluster, error) {
var err error
updaters := make(map[string]interface{})
if opt.Description != nil {
updaters["description"] = *opt.Description
defer func() {
if err == nil {
c.Description = *opt.Description
}
}()
}
if opt.KubeConfig != nil {
updaters["kube_config"] = *opt.KubeConfig
defer func() {
if err == nil {
c.KubeConfig = *opt.KubeConfig
}
}()
}
if len(updaters) == 0 {
return c, nil
}
db := s.getDB(ctx)
log.Info().Msgf("Updating cluster with updaters: %+v", updaters)
err = db.Where("id = ?", c.ID).Updates(updaters).Error
if err != nil {
log.Error().Msgf("Failed to update cluster: %s", err.Error())
return nil, err
}
return c, err
}
func (s *clusterService) Get(ctx context.Context, id uint) (*models.Cluster, error) {
var cluster models.Cluster
db := s.getDB(ctx)
err := db.Where("id = ?", id).First(&cluster).Error
if err != nil {
log.Error().Msgf("Failed to get cluster by id %d: %s", id, err.Error())
return nil, err
}
if cluster.ID == 0 {
return nil, consts.ErrNotFound
}
return &cluster, nil
}
func (s *clusterService) GetByUid(ctx context.Context, uid string) (*models.Cluster, error) {
var cluster models.Cluster
db := s.getDB(ctx)
err := db.Where("uid = ?", uid).First(&cluster).Error
if err != nil {
log.Error().Msgf("Failed to get cluster by uid %s: %s", uid, err.Error())
return nil, err
}
if cluster.ID == 0 {
return nil, consts.ErrNotFound
}
return &cluster, nil
}
func (s *clusterService) GetByName(ctx context.Context, organizationId string, name string) (*models.Cluster, error) {
var cluster models.Cluster
db := s.getDB(ctx)
err := db.Where("organization_id = ?", organizationId).Where("name = ?", name).First(&cluster).Error
if err != nil {
log.Error().Msgf("Failed to get cluster by name %s: %s", name, err.Error())
return nil, err
}
if cluster.ID == 0 {
return nil, consts.ErrNotFound
}
return &cluster, nil
}
func (s *clusterService) GetIdByName(ctx context.Context, organizationId uint, name string) (uint, error) {
var cluster models.Cluster
db := s.getDB(ctx)
err := db.Select("id").Where("organization_id = ?", organizationId).Where("name = ?", name).First(&cluster).Error
return cluster.ID, err
}
func (s *clusterService) List(ctx context.Context, opt ListClusterOption) ([]*models.Cluster, uint, error) {
clusters := make([]*models.Cluster, 0)
query := s.getDB(ctx)
if opt.Ids != nil {
if len(*opt.Ids) == 0 {
return clusters, 0, nil
}
query = query.Where("id in (?)", *opt.Ids)
}
if opt.Names != nil {
if len(*opt.Names) == 0 {
return clusters, 0, nil
}
query = query.Where("name in (?)", *opt.Names)
}
if opt.OrganizationId != nil {
query = query.Where("organization_id = ?", *opt.OrganizationId)
}
var total int64
err := query.Count(&total).Error
if err != nil {
return nil, 0, err
}
query = opt.BindQueryWithLimit(query)
if opt.Ids == nil {
if opt.Order == nil {
query = query.Order("id DESC")
} else {
query = query.Order(*opt.Order)
}
}
err = query.Find(&clusters).Error
if err != nil {
return nil, 0, err
}
return clusters, uint(total), err
}
func (s *clusterService) getDB(ctx context.Context) *gorm.DB {
db := database.DatabaseUtil.GetDBSession(ctx).Model(&models.Cluster{})
return db
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 services
import (
"context"
"strings"
"time"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/common/consts"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/database"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/models"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/schemas"
"github.com/pkg/errors"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"k8s.io/apimachinery/pkg/util/validation"
)
type compoundComponentService struct{}
var CompoundComponentService = compoundComponentService{}
type CreateCompoundComponentOption struct {
CreatorId uint
OrganizationId uint
ClusterId uint
Name string
Description string
Version string
KubeNamespace string
Manifest *schemas.CompoundComponentManifestSchema
}
type UpdateCompoundComponentOption struct {
Description *string
Version *string
LatestInstalledAt **time.Time
LatestHeartbeatAt **time.Time
Manifest **schemas.CompoundComponentManifestSchema
}
func (s *compoundComponentService) Create(ctx context.Context, opt CreateCompoundComponentOption) (*models.CompoundComponent, error) {
errs := validation.IsDNS1035Label(opt.Name)
if len(errs) > 0 {
return nil, errors.New(strings.Join(errs, ";"))
}
errs = validation.IsDNS1035Label(opt.KubeNamespace)
if len(errs) > 0 {
return nil, errors.New(strings.Join(errs, ";"))
}
now := time.Now()
compoundComponent := models.CompoundComponent{
Resource: models.Resource{
Name: opt.Name,
},
ClusterAssociate: models.ClusterAssociate{
ClusterId: opt.ClusterId,
},
Description: opt.Description,
KubeNamespace: opt.KubeNamespace,
Manifest: opt.Manifest,
Version: opt.Version,
LatestInstalledAt: &now,
LatestHeartbeatAt: &now,
}
err := s.getDB(ctx).Create(&compoundComponent).Error
if err != nil {
return nil, err
}
return &compoundComponent, err
}
func (s *compoundComponentService) Update(ctx context.Context, b *models.CompoundComponent, opt UpdateCompoundComponentOption) (*models.CompoundComponent, error) {
var err error
updaters := make(map[string]interface{})
if opt.Description != nil {
updaters["description"] = *opt.Description
defer func() {
if err == nil {
b.Description = *opt.Description
}
}()
}
if opt.LatestHeartbeatAt != nil {
updaters["latest_heartbeat_at"] = *opt.LatestHeartbeatAt
defer func() {
if err == nil {
b.LatestHeartbeatAt = *opt.LatestHeartbeatAt
}
}()
}
if opt.LatestInstalledAt != nil {
updaters["latest_installed_at"] = *opt.LatestInstalledAt
defer func() {
if err == nil {
b.LatestInstalledAt = *opt.LatestInstalledAt
}
}()
}
if opt.Version != nil {
updaters["version"] = *opt.Version
defer func() {
if err == nil {
b.Version = *opt.Version
}
}()
}
if opt.Manifest != nil {
updaters["manifest"] = *opt.Manifest
defer func() {
if err == nil {
b.Manifest = *opt.Manifest
}
}()
}
if len(updaters) == 0 {
return b, nil
}
err = s.getDB(ctx).Where("id = ?", b.ID).Updates(updaters).Error
if err != nil {
return nil, err
}
return b, err
}
func (s *compoundComponentService) Get(ctx context.Context, id uint) (*models.CompoundComponent, error) {
var compoundComponent models.CompoundComponent
err := s.getDB(ctx).Preload(clause.Associations).Where("id = ?", id).First(&compoundComponent).Error
if err != nil {
return nil, err
}
if compoundComponent.ID == 0 {
return nil, consts.ErrNotFound
}
return &compoundComponent, nil
}
func (s *compoundComponentService) GetByUid(ctx context.Context, uid string) (*models.CompoundComponent, error) {
var compoundComponent models.CompoundComponent
err := s.getDB(ctx).Preload(clause.Associations).Where("uid = ?", uid).First(&compoundComponent).Error
if err != nil {
return nil, err
}
if compoundComponent.ID == 0 {
return nil, consts.ErrNotFound
}
return &compoundComponent, nil
}
func (s *compoundComponentService) GetByName(ctx context.Context, clusterId uint, name string) (*models.CompoundComponent, error) {
var compoundComponent models.CompoundComponent
err := s.getDB(ctx).Where("cluster_id = ?", clusterId).Where("name = ?", name).First(&compoundComponent).Error
if err != nil {
return nil, errors.Wrapf(err, "get compoundComponent %s", name)
}
if compoundComponent.ID == 0 {
return nil, consts.ErrNotFound
}
return &compoundComponent, nil
}
func (s *compoundComponentService) ListByUids(ctx context.Context, uids []string) ([]*models.CompoundComponent, error) {
compoundComponents := make([]*models.CompoundComponent, 0, len(uids))
if len(uids) == 0 {
return compoundComponents, nil
}
err := s.getDB(ctx).Preload(clause.Associations).Where("uid in (?)", uids).Find(&compoundComponents).Error
return compoundComponents, err
}
type ListCompoundComponentOption struct {
Ids *[]uint `json:"ids"`
ClusterId *uint `json:"cluster_id"`
ClusterIds *[]uint `json:"cluster_ids"`
OrganizationId *uint `json:"organization_id"`
}
func (s *compoundComponentService) List(ctx context.Context, opt ListCompoundComponentOption) ([]*models.CompoundComponent, error) {
query := s.getDB(ctx).Preload(clause.Associations)
if opt.OrganizationId != nil {
query = query.Where("organization_id = ?", *opt.OrganizationId)
}
if opt.ClusterIds != nil {
query = query.Where("cluster_id in (?)", *opt.ClusterIds)
}
if opt.ClusterId != nil {
query = query.Where("cluster_id = ?", *opt.ClusterId)
}
if opt.Ids != nil {
query = query.Where("id in (?)", *opt.Ids)
}
compoundComponents := make([]*models.CompoundComponent, 0)
err := query.Find(&compoundComponents).Error
if err != nil {
return nil, err
}
return compoundComponents, err
}
func (s *compoundComponentService) getDB(ctx context.Context) *gorm.DB {
db := database.DatabaseUtil.GetDBSession(ctx).Model(&models.CompoundComponent{})
return db
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 services
import (
"context"
"encoding/json"
"fmt"
"net/http"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/common/client"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/common/env"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/schemas"
"github.com/rs/zerolog/log"
)
type datastoreService struct{}
var DatastoreService = datastoreService{}
/**
This service connects to the Nemo Datastore Microservice
Note: We should not do any write requests via this service as transactionality is not guaranteed in this way
**/
func (s *datastoreService) GetCompoundNimVersion(ctx context.Context, compoundNim string, compoundNimVersion string) (*schemas.CompoundNimVersionFullSchema, error) {
ndsUrl := env.GetNdsUrl()
getUrl := fmt.Sprintf("%s/api/v1/bento_repositories/%s/bentos/%s", ndsUrl, compoundNim, compoundNimVersion)
_, body, err := client.SendRequestJSON(getUrl, http.MethodGet, nil)
if err != nil {
log.Error().Msgf("Failed to get Compound NIM version %s:%s from %s", compoundNim, compoundNimVersion, ndsUrl)
return nil, err
}
var schema schemas.CompoundNimVersionFullSchema
if err = json.Unmarshal(body, &schema); err != nil {
log.Error().Msgf("Failed to unmarshal into a Compound NIM version schema: %s", err.Error())
return nil, err
}
return &schema, nil
}
/*
* SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 services
import (
"context"
"fmt"
"strings"
"time"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/common/consts"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/database"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/models"
"github.com/dynemo-ai/dynemo/deploy/compoundai/api-server/api/schemas"
"github.com/google/uuid"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
"gorm.io/gorm"
apiv1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/util/validation"
)
type deploymentService struct{}
var DeploymentService = deploymentService{}
type CreateDeploymentOption struct {
CreatorId string
ClusterId uint
Name string
Description string
KubeNamespace string
}
type UpdateDeploymentOption struct {
Description *string
Status *schemas.DeploymentStatus
}
type UpdateDeploymentStatusOption struct {
Status *schemas.DeploymentStatus
SyncingAt **time.Time
UpdatedAt **time.Time
}
type ListDeploymentOption struct {
BaseListOption
ClusterId *uint
CreatorId *string
LastUpdaterId *uint
OrganizationId *string
ClusterIds *[]string
CreatorIds *[]uint
LastUpdaterIds *[]uint
OrganizationIds *[]string
Ids *[]uint
CompoundNimVersionIds *[]uint
Statuses *[]schemas.DeploymentStatus
Order *string
CompoundNimName *string
CompoundNimTag *string
}
func (s *deploymentService) Create(ctx context.Context, opt CreateDeploymentOption) (*models.Deployment, error) {
errs := validation.IsDNS1035Label(opt.Name)
if len(errs) > 0 {
return nil, errors.New(strings.Join(errs, ";"))
}
errs = validation.IsDNS1035Label(opt.KubeNamespace)
if len(errs) > 0 {
return nil, errors.New(strings.Join(errs, ";"))
}
guid := uuid.New()
deployment := models.Deployment{
Resource: models.Resource{
Name: opt.Name,
},
ClusterAssociate: models.ClusterAssociate{
ClusterId: opt.ClusterId,
},
CreatorAssociate: models.CreatorAssociate{
UserId: opt.CreatorId,
},
Description: opt.Description,
Status: schemas.DeploymentStatusNonDeployed,
KubeDeployToken: guid.String(),
KubeNamespace: opt.KubeNamespace,
}
db := s.getDB(ctx)
err := db.Create(&deployment).Error
if err != nil {
log.Error().Msgf("Failed to create deployment %s", err.Error())
return nil, err
}
return &deployment, err
}
func (s *deploymentService) Update(ctx context.Context, b *models.Deployment, opt UpdateDeploymentOption) (*models.Deployment, error) {
var err error
updaters := make(map[string]interface{})
if opt.Description != nil {
updaters["description"] = *opt.Description
defer func() {
if err == nil {
b.Description = *opt.Description
}
}()
}
if opt.Status != nil {
updaters["status"] = *opt.Status
defer func() {
if err == nil {
b.Status = *opt.Status
}
}()
}
if len(updaters) == 0 {
return b, nil
}
log.Info().Msgf("Updating deployment with updaters %+v", updaters)
err = s.getDB(ctx).Where("id = ?", b.ID).Updates(updaters).Error
if err != nil {
return nil, err
}
return b, err
}
func (s *deploymentService) Get(ctx context.Context, id uint) (*models.Deployment, error) {
var deployment models.Deployment
err := s.getDB(ctx).Where("id = ?", id).First(&deployment).Error
if err != nil {
log.Error().Msgf("Failed to get deployment by id %d: %s", id, err.Error())
return nil, err
}
if deployment.ID == 0 {
return nil, consts.ErrNotFound
}
return &deployment, nil
}
func (s *deploymentService) GetByUid(ctx context.Context, uid string) (*models.Deployment, error) {
var deployment models.Deployment
err := s.getDB(ctx).Where("uid = ?", uid).First(&deployment).Error
if err != nil {
log.Error().Msgf("Failed to get deployment by uid %s: %s", uid, err.Error())
return nil, err
}
if deployment.ID == 0 {
return nil, consts.ErrNotFound
}
return &deployment, nil
}
func (s *deploymentService) GetByName(ctx context.Context, clusterId uint, kubeNamespace, name string) (*models.Deployment, error) {
var deployment models.Deployment
err := s.getDB(ctx).Where("cluster_id = ?", clusterId).Where("kube_namespace = ?", kubeNamespace).Where("name = ?", name).First(&deployment).Error
if err != nil {
log.Error().Msgf("Failed to get deployment by name and creator %s: %s", name, err.Error())
return nil, err
}
if deployment.ID == 0 {
return nil, consts.ErrNotFound
}
return &deployment, nil
}
func (s *deploymentService) GetByNameAndCreator(ctx context.Context, clusterId uint, kubeNamespace, name string, creatorId string) (*models.Deployment, error) {
var deployment models.Deployment
err := s.getDB(ctx).Where("cluster_id = ?", clusterId).Where("kube_namespace = ?", kubeNamespace).Where("name = ?", name).Where("user_id = ?", creatorId).First(&deployment).Error
if err != nil {
log.Error().Msgf("Failed to get deployment by name %s: %s", name, err.Error())
return nil, err
}
if deployment.ID == 0 {
return nil, consts.ErrNotFound
}
return &deployment, nil
}
func (s *deploymentService) Delete(ctx context.Context, deployment *models.Deployment) (*models.Deployment, error) {
if deployment.Status != schemas.DeploymentStatusTerminated && deployment.Status != schemas.DeploymentStatusTerminating {
return nil, errors.New("deployment is not terminated")
}
return deployment, s.getDB(ctx).Unscoped().Delete(deployment).Error
}
func (s *deploymentService) Terminate(ctx context.Context, deployment *models.Deployment) (*models.Deployment, error) {
deployment, err := s.UpdateStatus(ctx, deployment, UpdateDeploymentStatusOption{
Status: schemas.DeploymentStatusTerminating.Ptr(),
})
if err != nil {
return nil, err
}
start := uint(0)
count := uint(1)
deploymentRevisions, _, err := DeploymentRevisionService.List(ctx, ListDeploymentRevisionOption{
BaseListOption: BaseListOption{
Start: &start,
Count: &count,
},
DeploymentId: &deployment.ID,
Status: schemas.DeploymentRevisionStatusActive.Ptr(),
})
if err != nil {
return nil, err
}
log.Info().Msgf("Fetched %d active deployment revisions to terminate", len(deploymentRevisions))
for _, deploymentRevision := range deploymentRevisions {
err = DeploymentRevisionService.Terminate(ctx, deploymentRevision)
if err != nil {
return nil, err
}
}
_, err = s.SyncStatus(ctx, deployment)
return deployment, err
}
func (s *deploymentService) UpdateStatus(ctx context.Context, deployment *models.Deployment, opt UpdateDeploymentStatusOption) (*models.Deployment, error) {
updater := map[string]interface{}{}
if opt.Status != nil {
deployment.Status = *opt.Status
updater["status"] = *opt.Status
}
if opt.SyncingAt != nil {
deployment.StatusSyncingAt = *opt.SyncingAt
updater["status_syncing_at"] = *opt.SyncingAt
}
if opt.UpdatedAt != nil {
deployment.StatusUpdatedAt = *opt.UpdatedAt
updater["status_updated_at"] = *opt.UpdatedAt
}
log.Info().Msgf("Updating deployment with updaters %+v", updater)
err := s.getDB(ctx).Where("id = ?", deployment.ID).Updates(updater).Error
return deployment, err
}
func (s *deploymentService) SyncStatus(ctx context.Context, d *models.Deployment) (schemas.DeploymentStatus, error) {
now := time.Now()
nowPtr := &now
_, err := s.UpdateStatus(ctx, d, UpdateDeploymentStatusOption{
SyncingAt: &nowPtr,
})
if err != nil {
log.Error().Msgf("Failed to update sync time for deployment %s: %s", d.Name, err.Error())
return d.Status, err
}
currentStatus, err := s.getStatusFromK8s(ctx, d)
if err != nil {
log.Error().Msgf("Failed to get deployment status from k8s for deployment %s: %s", d.Name, err.Error())
return currentStatus, err
}
now = time.Now()
nowPtr = &now
_, err = s.UpdateStatus(ctx, d, UpdateDeploymentStatusOption{
Status: &currentStatus,
UpdatedAt: &nowPtr,
})
if err != nil {
return currentStatus, err
}
return currentStatus, nil
}
func (s *deploymentService) List(ctx context.Context, opt ListDeploymentOption) ([]*models.Deployment, uint, error) {
query := s.getDB(ctx)
if opt.Ids != nil {
query = query.Where("deployment.id in (?)", *opt.Ids)
}
query = query.Joins("LEFT JOIN deployment_revision ON deployment_revision.deployment_id = deployment.id AND deployment_revision.status = ?", schemas.DeploymentRevisionStatusActive)
joinOnDeploymentTargets := query.Joins("LEFT JOIN deployment_target ON deployment_target.deployment_revision_id = deployment_revision.id")
if opt.CompoundNimName != nil {
query = joinOnDeploymentTargets.Where("deployment_target.compound_nim_version_tag LIKE ?", *opt.CompoundNimName+":%")
}
if opt.CompoundNimTag != nil {
query = joinOnDeploymentTargets.Where("deployment_target.compound_nim_version_tag = ?", *opt.CompoundNimTag)
}
if opt.CompoundNimVersionIds != nil {
query = joinOnDeploymentTargets.Where("deployment_target.compound_nim_version_id IN (?)", *opt.CompoundNimVersionIds)
}
if opt.ClusterId != nil {
query = query.Where("deployment.cluster_id = ?", *opt.ClusterId)
}
if opt.ClusterIds != nil {
query = query.Where("deployment.cluster_id IN (?)", *opt.ClusterIds)
}
if opt.Statuses != nil {
query = query.Where("deployment.status IN (?)", *opt.Statuses)
}
if opt.OrganizationId != nil {
query = query.Joins("LEFT JOIN cluster ON cluster.id = deployment.cluster_id")
query = query.Where("cluster.organization_id = ?", *opt.OrganizationId)
}
if opt.CreatorId != nil {
query = query.Where("deployment.user_id = ?", *opt.CreatorId)
}
query = opt.BindQueryWithKeywords(query, "deployment")
query = query.Select("deployment_revision.*, deployment.*")
var total int64
err := query.Count(&total).Error
if err != nil {
return nil, 0, err
}
query = opt.BindQueryWithLimit(query)
if opt.Order != nil {
query = query.Order(*opt.Order)
} else {
query.Order("deployment.id DESC")
}
deployments := make([]*models.Deployment, 0)
err = query.Find(&deployments).Error
if err != nil {
return nil, 0, err
}
return deployments, uint(total), err
}
func (s *deploymentService) getDB(ctx context.Context) *gorm.DB {
db := database.DatabaseUtil.GetDBSession(ctx).Model(&models.Deployment{})
return db
}
func (s *deploymentService) getStatusFromK8s(ctx context.Context, d *models.Deployment) (schemas.DeploymentStatus, error) {
defaultStatus := schemas.DeploymentStatusUnknown
cluster, err := ClusterService.Get(ctx, d.ClusterId)
if err != nil {
return defaultStatus, err
}
namespace := d.KubeNamespace
_, podLister, err := GetPodInformer(ctx, cluster, namespace)
if err != nil {
return defaultStatus, err
}
imageBuilderPods := make([]*apiv1.Pod, 0)
status_ := schemas.DeploymentRevisionStatusActive
deploymentRevisions, _, err := DeploymentRevisionService.List(ctx, ListDeploymentRevisionOption{
DeploymentId: &d.ID,
Status: &status_,
})
if err != nil {
return defaultStatus, err
}
deploymentRevisionIds := make([]uint, 0, len(deploymentRevisions))
for _, deploymentRevision := range deploymentRevisions {
deploymentRevisionIds = append(deploymentRevisionIds, deploymentRevision.ID)
}
deploymentTargets, _, err := DeploymentTargetService.List(ctx, ListDeploymentTargetOption{
DeploymentRevisionIds: &deploymentRevisionIds,
})
if err != nil {
return defaultStatus, err
}
for _, deploymentTarget := range deploymentTargets {
compoundNimParts := strings.Split(deploymentTarget.CompoundNimVersionTag, ":")
if len(compoundNimParts) != 2 {
return defaultStatus, errors.Errorf("Invalid format for CompoundNIM version tag %s. Expected 2 parts got %d", deploymentTarget.CompoundNimVersionTag, len(compoundNimParts))
}
imageBuilderPodsSelector, err := labels.Parse(fmt.Sprintf("%s=%s,%s=%s", consts.KubeLabelCompoundNim, compoundNimParts[0], consts.KubeLabelCompoundNimVersion, compoundNimParts[1]))
if err != nil {
return defaultStatus, err
}
var pods_ []*apiv1.Pod
pods_, err = K8sService.ListPodsBySelector(ctx, podLister, imageBuilderPodsSelector)
if err != nil {
return defaultStatus, err
}
imageBuilderPods = append(imageBuilderPods, pods_...)
}
log.Info().Msgf("Fetched %d image builder jobs", len(imageBuilderPods))
if len(imageBuilderPods) != 0 {
for _, imageBuilderPod := range imageBuilderPods {
for _, container := range imageBuilderPod.Status.ContainerStatuses {
if container.Name == consts.KubeImageBuilderMainContainer {
if container.State.Waiting != nil || container.State.Running != nil {
return schemas.DeploymentStatusImageBuilding, nil
} else if container.State.Terminated != nil {
if container.State.Terminated.ExitCode != 0 {
return schemas.DeploymentStatusImageBuildFailed, nil
}
}
}
}
}
}
pods, err := K8sService.ListPodsByDeployment(ctx, podLister, d)
if err != nil {
return defaultStatus, err
}
log.Info().Msgf("Fetched %d pods", len(pods))
if len(pods) == 0 {
if d.Status == schemas.DeploymentStatusTerminating || d.Status == schemas.DeploymentStatusTerminated {
return schemas.DeploymentStatusTerminated, nil
}
if d.Status == schemas.DeploymentStatusDeploying {
return schemas.DeploymentStatusDeploying, nil
}
return schemas.DeploymentStatusNonDeployed, nil
}
if d.Status == schemas.DeploymentStatusTerminated {
return d.Status, nil
}
hasFailed := false
hasRunning := false
hasPending := false
for _, p := range pods {
log.Info().Msgf("pod %s has status %s", p.Name, p.Status.Phase)
podStatus := p.Status
if podStatus.Phase == apiv1.PodRunning {
hasRunning = true
}
if podStatus.Phase == apiv1.PodFailed {
hasFailed = true
}
if podStatus.Phase == apiv1.PodPending {
hasPending = true
}
}
var deploymentStatus schemas.DeploymentStatus
if d.Status == schemas.DeploymentStatusTerminating {
if !hasRunning {
deploymentStatus = schemas.DeploymentStatusTerminated
} else {
deploymentStatus = schemas.DeploymentStatusTerminating
}
} else if hasFailed && hasRunning {
if hasPending {
deploymentStatus = schemas.DeploymentStatusDeploying
} else {
deploymentStatus = schemas.DeploymentStatusUnhealthy
}
} else if hasPending {
deploymentStatus = schemas.DeploymentStatusDeploying
} else if hasRunning {
deploymentStatus = schemas.DeploymentStatusRunning
}
log.Info().Msgf("The current status of the deployment is %s", deploymentStatus)
return deploymentStatus, nil
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment