manager.go 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. package plugin_manager
  2. import (
  3. "fmt"
  4. "github.com/langgenius/dify-plugin-daemon/internal/cluster"
  5. "github.com/langgenius/dify-plugin-daemon/internal/core/dify_invocation"
  6. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/media_manager"
  7. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/serverless"
  8. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_packager/decoder"
  9. "github.com/langgenius/dify-plugin-daemon/internal/types/app"
  10. "github.com/langgenius/dify-plugin-daemon/internal/types/entities/plugin_entities"
  11. "github.com/langgenius/dify-plugin-daemon/internal/utils/cache"
  12. "github.com/langgenius/dify-plugin-daemon/internal/utils/lock"
  13. "github.com/langgenius/dify-plugin-daemon/internal/utils/log"
  14. "github.com/langgenius/dify-plugin-daemon/internal/utils/mapping"
  15. "github.com/langgenius/dify-plugin-daemon/internal/utils/stream"
  16. )
  17. type PluginManager struct {
  18. m mapping.Map[string, plugin_entities.PluginLifetime]
  19. cluster *cluster.Cluster
  20. maxPluginPackageSize int64
  21. workingDirectory string
  22. // mediaManager is used to manage media files like plugin icons, images, etc.
  23. mediaManager *media_manager.MediaManager
  24. // running plugin in storage contains relations between plugin packages and their running instances
  25. runningPluginInStorage mapping.Map[string, string]
  26. // start process lock
  27. startProcessLock *lock.HighGranularityLock
  28. // serverless runtime
  29. // Install is a function that installs a plugin to the platform
  30. Install func(decoder decoder.PluginDecoder) (*stream.Stream[PluginInstallResponse], error)
  31. }
  32. var (
  33. manager *PluginManager
  34. )
  35. func InitManager(cluster *cluster.Cluster, configuration *app.Config) {
  36. manager = &PluginManager{
  37. cluster: cluster,
  38. maxPluginPackageSize: configuration.MaxPluginPackageSize,
  39. workingDirectory: configuration.PluginWorkingPath,
  40. mediaManager: media_manager.NewMediaManager(
  41. configuration.PluginMediaCachePath,
  42. configuration.PluginMediaCacheSize,
  43. ),
  44. startProcessLock: lock.NewHighGranularityLock(),
  45. }
  46. if configuration.Platform == app.PLATFORM_AWS_LAMBDA {
  47. manager.Install = manager.InstallToAWSFromPkg
  48. serverless.Init(configuration)
  49. } else if configuration.Platform == app.PLATFORM_LOCAL {
  50. manager.Install = manager.InstallToLocal
  51. }
  52. manager.Init(configuration)
  53. }
  54. func Manager() *PluginManager {
  55. return manager
  56. }
  57. func (p *PluginManager) Add(
  58. plugin plugin_entities.PluginLifetime,
  59. ) error {
  60. identity, err := plugin.Identity()
  61. if err != nil {
  62. return err
  63. }
  64. p.m.Store(identity.String(), plugin)
  65. return nil
  66. }
  67. func (p *PluginManager) Get(
  68. identity plugin_entities.PluginUniqueIdentifier,
  69. ) plugin_entities.PluginLifetime {
  70. if v, ok := p.m.Load(identity.String()); ok {
  71. return v
  72. }
  73. // check if plugin is a serverless runtime
  74. plugin_session_interface, err := p.getServerlessPluginRuntime(identity)
  75. if err != nil {
  76. return nil
  77. }
  78. return plugin_session_interface
  79. }
  80. func (p *PluginManager) GetAsset(id string) ([]byte, error) {
  81. return p.mediaManager.Get(id)
  82. }
  83. func (p *PluginManager) Init(configuration *app.Config) {
  84. // TODO: init plugin manager
  85. log.Info("start plugin manager daemon...")
  86. // init redis client
  87. if err := cache.InitRedisClient(
  88. fmt.Sprintf("%s:%d", configuration.RedisHost, configuration.RedisPort),
  89. configuration.RedisPass,
  90. ); err != nil {
  91. log.Panic("init redis client failed: %s", err.Error())
  92. }
  93. if err := dify_invocation.InitDifyInvocationDaemon(
  94. configuration.PluginInnerApiURL, configuration.PluginInnerApiKey,
  95. ); err != nil {
  96. log.Panic("init dify invocation daemon failed: %s", err.Error())
  97. }
  98. // start local watcher
  99. if configuration.Platform == app.PLATFORM_LOCAL {
  100. p.startLocalWatcher(configuration)
  101. }
  102. // start remote watcher
  103. p.startRemoteWatcher(configuration)
  104. }