Files
Yeuoly 888ad788bc refactor: plugin lifecycle control panel (#499)
* refactor: introduce local plugin control panel and cleanup environment setup process

* fix: args

* refactor: new local runtime

* temp: stash work for refactor on RemotePluginServer

* refactor: unify local runtime lifetime and sperate init environment process

* chore: add missing files

* stash

* refactor: local plugin lifetime control

* refactor: complete installation process of control panel

* refactor: adapt service layer to new controlpanel

* refactor: pluginManager.Install

* fix: add routine wrap to InstallServerless, avoid blocking main thread

* feat: reinstall serverless runtime

* chore: add comments to Reinstall and update confusing naming

* refactor: unify install plugin service

* refactor: add labels to debugging runtime

* refactor: add getters to plugin manager

* refactor: split install service to decode/install_task/install service

* ???

* refactor: adapt controllers

* refactor: session write

* refactor: session runtime

* Refine install task orchestration (#501)

* refactor: installing task

* refactor cluster management, decouple lifetime management and cluster

* fix cli test command

* fix: cleanup TODO comments and implement GracefulStop for instance

* feat: add logger to control panel

* fix: multiple nil references

* refactor: better lifetime control

* refactor: better cycle interval

* fix(LocalPluginRuntime): prevent returning err when it's not  error

* fix: avoid adding empty PipExtraArgs

* fix: missing errors in Environment init

* fix: add truncateMessage to avoid db explosion

* cleanup: better lifecycle management

* fix: init status at the beginning of installation

* optimize: GracefulStop for pluginInstance

* refactor: tests

* refactor: centralize routine labels (#504)

* cleanup: RoutineKey

* fix: init routine pool

* fix: correctly handle cluster register error

* fix: memory leak

* fix: add \n to instance write

* fix(installer.go): set success to true after succeed for defer func

* refactor

* fix: missing cwd in testutils

* fix: scaleup default runtime nums to 1 when testing

* fix: localruntime appconfig in testing module

* Update internal/core/local_runtime/load_balancing.go

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* fix: more efficiency implement in installer_local.go

* fix: returns after failing in onDebuggingRuntimeDisconnected

* fix: returns after failing in onDebuggingRuntimeDisconnected

* fix: splits tests

* refactor: naming

* refactor: manifest.VersionX

* fix: adapt SetDefault to tests

* fix: enforce use constants in DBType

* fix: generate

* fix: linter

* cleanup tests

* refactor: change  package to

* cleanup: useless codes

* Update internal/cluster/plugin.go

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* cleanup

* refactor: decouple connection_key management from debugging_time

* refactor: confused naming

* feat: recycle resources to adapt to https://github.com/langgenius/dify-plugin-daemon/pull/500

* refactor: confusing redirecting

* fix: support get serverless runtime

* fix: race condition in Launching

* fix: avoid ManifestValidate in first step of debugging handshake

* fix: adding ReleaseAllLocks to finalizers

* wtf: what a beautiful code

* refactor: rename Stream.Async to Stream.Process

* fix: kill process if daed instance was detected

* fix: correctly handle failures

* fix: consistence of difference interfaces

* fix: add stacktrace to panic

* fix: only trigger once  event

* fix: ensure plugin runtime was shutdown

* feat: cleanup install tasks

* fix: add scale logs

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
2025-11-18 17:28:02 +08:00

177 lines
4.0 KiB
Go

package persistence
import (
"encoding/hex"
"fmt"
"path"
"strings"
"time"
"github.com/langgenius/dify-plugin-daemon/internal/db"
"github.com/langgenius/dify-plugin-daemon/internal/types/models"
"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
)
type Persistence struct {
maxStorageSize int64
storage PersistenceStorage
}
const (
CACHE_KEY_PREFIX = "persistence:cache"
)
func (c *Persistence) getCacheKey(tenantId string, pluginId string, key string) string {
return fmt.Sprintf("%s:%s:%s:%s", CACHE_KEY_PREFIX, tenantId, pluginId, key)
}
func (c *Persistence) checkPathTraversal(key string) error {
key = path.Clean(key)
if strings.Contains(key, "..") || strings.Contains(key, "//") || strings.Contains(key, "\\") {
return fmt.Errorf("invalid key: path traversal attempt detected")
}
return nil
}
func (c *Persistence) Save(tenantId string, pluginId string, maxSize int64, key string, data []byte) error {
if err := c.checkPathTraversal(key); err != nil {
return err
}
if len(key) > 256 {
return fmt.Errorf("key length must be less than 256 characters")
}
if maxSize == -1 {
maxSize = c.maxStorageSize
}
if err := c.storage.Save(tenantId, pluginId, key, data); err != nil {
return err
}
allocatedSize := int64(len(data))
storage, err := db.GetOne[models.TenantStorage](
db.Equal("tenant_id", tenantId),
db.Equal("plugin_id", pluginId),
)
if err != nil {
if allocatedSize > c.maxStorageSize || allocatedSize > maxSize {
return fmt.Errorf("allocated size is greater than max storage size")
}
if err == db.ErrDatabaseNotFound {
storage = models.TenantStorage{
TenantID: tenantId,
PluginID: pluginId,
Size: allocatedSize,
}
if err := db.Create(&storage); err != nil {
return err
}
} else {
return err
}
} else {
if allocatedSize+storage.Size > maxSize || allocatedSize+storage.Size > c.maxStorageSize {
return fmt.Errorf("allocated size is greater than max storage size")
}
err = db.Run(
db.Model(&models.TenantStorage{}),
db.Equal("tenant_id", tenantId),
db.Equal("plugin_id", pluginId),
db.Inc(map[string]int64{"size": allocatedSize}),
)
if err != nil {
return err
}
}
// delete from cache
if _, err = cache.Del(c.getCacheKey(tenantId, pluginId, key)); err == cache.ErrNotFound {
return nil
}
return err
}
// TODO: raises specific error to avoid confusion
func (c *Persistence) Load(tenantId string, pluginId string, key string) ([]byte, error) {
if err := c.checkPathTraversal(key); err != nil {
return nil, err
}
// check if the key exists in cache
h, err := cache.GetString(c.getCacheKey(tenantId, pluginId, key))
if err != nil && err != cache.ErrNotFound {
return nil, err
}
if err == nil {
return hex.DecodeString(h)
}
// load from storage
data, err := c.storage.Load(tenantId, pluginId, key)
if err != nil {
return nil, err
}
// add to cache
cache.Store(c.getCacheKey(tenantId, pluginId, key), hex.EncodeToString(data), time.Minute*5)
return data, nil
}
func (c *Persistence) Delete(tenantId string, pluginId string, key string) (int64, error) {
// delete from cache and storage
deletedNum, err := cache.Del(c.getCacheKey(tenantId, pluginId, key))
if err != nil {
return 0, err
}
// state size
size, err := c.storage.StateSize(tenantId, pluginId, key)
if err != nil {
return 0, err
}
err = c.storage.Delete(tenantId, pluginId, key)
if err != nil {
return 0, err
}
// update storage size
err = db.Run(
db.Model(&models.TenantStorage{}),
db.Equal("tenant_id", tenantId),
db.Equal("plugin_id", pluginId),
db.Dec(map[string]int64{"size": size}),
)
if err != nil {
return 0, err
}
return deletedNum, nil
}
func (c *Persistence) Exist(tenantId string, pluginId string, key string) (int64, error) {
existNum, err := cache.Exist(c.getCacheKey(tenantId, pluginId, key))
if err != nil {
return 0, err
}
if existNum > 0 {
return existNum, nil
}
isExist, err := c.storage.Exists(tenantId, pluginId, key)
if err != nil {
return 0, err
}
if isExist {
return 1, nil
}
return 0, nil
}