dify-plugin-daemon/internal/core/plugin_manager/manager.go

package plugin_manager

import (
	"fmt"

	lru "github.com/hashicorp/golang-lru/v2"
	"github.com/langgenius/dify-cloud-kit/oss"
	"github.com/langgenius/dify-plugin-daemon/internal/cluster"
	controlpanel "github.com/langgenius/dify-plugin-daemon/internal/core/control_panel"
	"github.com/langgenius/dify-plugin-daemon/internal/core/dify_invocation"
	"github.com/langgenius/dify-plugin-daemon/internal/core/dify_invocation/calldify"
	"github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/media_transport"
	serverless "github.com/langgenius/dify-plugin-daemon/internal/core/serverless_connector"
	"github.com/langgenius/dify-plugin-daemon/internal/service/install_service"
	"github.com/langgenius/dify-plugin-daemon/internal/types/app"
	"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
	"github.com/langgenius/dify-plugin-daemon/pkg/plugin_packager/decoder"
	"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
	"github.com/langgenius/dify-plugin-daemon/pkg/utils/log"
	"github.com/langgenius/dify-plugin-daemon/pkg/utils/parser"
)

type PluginManager struct {
	// mediaBucket is used to manage media files like plugin icons, images, etc.
	mediaBucket *media_transport.MediaBucket

	// packageBucket can be considered as a collection of all uploaded plugin
	// original packages, once a package was accepted by Dify,
	// it should be stored here.
	packageBucket *media_transport.PackageBucket

	// installedBucket is used to manage installed plugins, all the installed plugins will be saved here
	// `accepted` dose not means `installed`, a installed plugin
	// will be scheduled by daemon, daemon copied and move the package
	// from `packageBucket` to `installedBucket`
	// the copy processing marks a plugin as `installed`
	// as for `scheduling`, it's automatically done by control panel
	// of course you may use `controlPanel.LaunchLocalPlugin` to start it manually
	installedBucket *media_transport.InstalledBucket

	// backwardsInvocation is a handle to invoke dify
	backwardsInvocation dify_invocation.BackwardsInvocation

	config *app.Config

	pluginAssetCache *lru.Cache[string, []byte]

	// plugin lifecycle controller
	//
	// whatever it's local mode or serverless mode, all the signals and calls
	// which related to plugin lifecycle should be handled by it.
	// so that we can decouple lifetime control and thirdparty service like package management
	controlPanel *controlpanel.ControlPanel
}

var (
	manager *PluginManager
)

func InitGlobalManager(oss oss.OSS, config *app.Config) *PluginManager {
	mediaBucket := media_transport.NewAssetsBucket(
		oss,
		config.PluginMediaCachePath,
		config.PluginMediaCacheSize,
	)

	installedBucket := media_transport.NewInstalledBucket(
		oss,
		config.PluginInstalledPath,
	)

	packageBucket := media_transport.NewPackageBucket(
		oss,
		config.PluginPackageCachePath,
	)

	pluginAssetCache, err := lru.New[string, []byte](int(config.PluginAssetCacheSize))
	if err != nil {
		log.Panic("init plugin asset cache failed", "error", err)
	}

	manager = &PluginManager{
		mediaBucket:      mediaBucket,
		packageBucket:    packageBucket,
		installedBucket:  installedBucket,
		pluginAssetCache: pluginAssetCache,
		controlPanel: controlpanel.NewControlPanel(
			config,
			mediaBucket,
			packageBucket,
			installedBucket,
			nil, // cluster will be set later via SetCluster
		),
		config: config,
	}

	// mount control panel notifiers
	manager.controlPanel.AddNotifier(&controlpanel.StandardLogger{})
	manager.controlPanel.AddNotifier(&install_service.InstallListener{})
	manager.controlPanel.AddNotifier(controlpanel.NewMetricsNotifier())

	return manager
}

func (p *PluginManager) SetCluster(cluster *cluster.Cluster) {
	p.controlPanel.SetCluster(cluster)
}

func Manager() *PluginManager {
	return manager
}

func (p *PluginManager) GetAsset(id string) ([]byte, error) {
	return p.mediaBucket.Get(id)
}

func (p *PluginManager) Launch(configuration *app.Config) {
	log.Info("start plugin manager daemon")

	// Build TLS config for Redis (nil when RedisUseSsl=false)
	tlsConf, err := configuration.RedisTLSConfig()
	if err != nil {
		log.Panic("invalid Redis TLS config: %s", err.Error())
	}

	cache.SetKeyPrefix(configuration.RedisKeyPrefix)

	// init redis client
	if configuration.RedisUseSentinel {
		// use Redis Sentinel
		sentinels := parser.SplitAndTrimCSV(configuration.RedisSentinels)
		if err := cache.InitRedisSentinelClient(
			sentinels,
			configuration.RedisSentinelServiceName,
			configuration.RedisUser,
			configuration.RedisPass,
			configuration.RedisSentinelUsername,
			configuration.RedisSentinelPassword,
			configuration.RedisUseSsl,
			configuration.RedisDB,
			configuration.RedisSentinelSocketTimeout,
			tlsConf, // pass TLS to cache initializer
		); err != nil {
			log.Panic("init redis sentinel client failed", "error", err)
		}
	} else {
		if err := cache.InitRedisClient(
			fmt.Sprintf("%s:%d", configuration.RedisHost, configuration.RedisPort),
			configuration.RedisUser,
			configuration.RedisPass,
			configuration.RedisUseSsl,
			configuration.RedisDB,
			tlsConf, // pass TLS to cache initializer
		); err != nil {
			log.Panic("init redis client failed", "error", err)
		}
	}

	invocation, err := calldify.NewDifyInvocationDaemon(
		calldify.NewDifyInvocationDaemonPayload{
			BaseUrl:               configuration.DifyInnerApiURL,
			CallingKey:            configuration.DifyInnerApiKey,
			WriteTimeout:          configuration.DifyInvocationWriteTimeout,
			ReadTimeout:           configuration.DifyInvocationReadTimeout,
			ResponseMaxBufferSize: configuration.ResponseMaxBufferSize,
		},
	)
	if err != nil {
		log.Panic("init dify invocation daemon failed", "error", err)
	}
	p.backwardsInvocation = invocation

	// start control panel
	p.controlPanel.StartWatchDog()

	// launch serverless connector
	if configuration.Platform == app.PLATFORM_SERVERLESS {
		serverless.Init(configuration)
	}
}

func (p *PluginManager) BackwardsInvocation() dify_invocation.BackwardsInvocation {
	return p.backwardsInvocation
}

func (p *PluginManager) Config() *app.Config {
	return p.config
}

// check if the plugin is already running on this node
func (c *PluginManager) NeedRedirecting(
	identity plugin_entities.PluginUniqueIdentifier,
) (bool, error) {
	// debugging runtime were stored in control panel
	if identity.RemoteLike() {
		_, err := c.controlPanel.GetPluginRuntime(identity)
		if err != nil {
			return true, err
		}
		return false, nil
	}

	if c.config.Platform == app.PLATFORM_SERVERLESS {
		// under serverless mode, it's no need to do redirecting
		return false, nil
	} else if c.config.Platform == app.PLATFORM_LOCAL {
		// under local mode, check if the plugin is already running on this node
		_, err := c.controlPanel.GetPluginRuntime(identity)
		if err != nil {
			// not found on this node, need to redirecting
			return true, err
		}

		// found on current node
		return false, nil
	}

	return true, nil
}

func pluginAssetCacheKey(
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
	path string,
) string {
	return fmt.Sprintf("%s/%s", pluginUniqueIdentifier.String(), path)
}

func (p *PluginManager) ExtractPluginAsset(
	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
	path string,
) ([]byte, error) {
	key := pluginAssetCacheKey(pluginUniqueIdentifier, path)
	cached, ok := p.pluginAssetCache.Get(key)
	if ok {
		return cached, nil
	}
	pkgBytes, err := p.GetPackage(pluginUniqueIdentifier)
	if err != nil {
		return nil, err
	}
	zipDecoder, err := decoder.NewZipPluginDecoder(pkgBytes)
	if err != nil {
		return nil, err
	}
	assets, err := zipDecoder.Assets()
	if err != nil {
		return nil, err
	}
	p.pluginAssetCache.Add(key, assets[path])
	return assets[path], nil
}