watcher.go 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212
  1. package plugin_manager
  2. import (
  3. "errors"
  4. "fmt"
  5. "io"
  6. "os"
  7. "path"
  8. "strings"
  9. "time"
  10. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/basic_manager"
  11. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/local_manager"
  12. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/positive_manager"
  13. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/remote_manager"
  14. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_packager/decoder"
  15. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_packager/verifier"
  16. "github.com/langgenius/dify-plugin-daemon/internal/types/app"
  17. "github.com/langgenius/dify-plugin-daemon/internal/types/entities/plugin_entities"
  18. "github.com/langgenius/dify-plugin-daemon/internal/utils/log"
  19. "github.com/langgenius/dify-plugin-daemon/internal/utils/routine"
  20. )
  21. func (p *PluginManager) startLocalWatcher(config *app.Config) {
  22. go func() {
  23. log.Info("start to handle new plugins in path: %s", config.PluginStoragePath)
  24. p.handleNewLocalPlugins(config)
  25. for range time.NewTicker(time.Second * 30).C {
  26. p.handleNewLocalPlugins(config)
  27. }
  28. }()
  29. }
  30. func (p *PluginManager) startRemoteWatcher(config *app.Config) {
  31. // launch TCP debugging server if enabled
  32. if config.PluginRemoteInstallingEnabled {
  33. server := remote_manager.NewRemotePluginServer(config, p.mediaManager)
  34. go func() {
  35. err := server.Launch()
  36. if err != nil {
  37. log.Error("start remote plugin server failed: %s", err.Error())
  38. }
  39. }()
  40. go func() {
  41. server.Wrap(func(rpr *remote_manager.RemotePluginRuntime) {
  42. p.fullDuplexLifetime(rpr)
  43. })
  44. }()
  45. }
  46. }
  47. func (p *PluginManager) handleNewLocalPlugins(config *app.Config) {
  48. // load local plugins firstly
  49. for plugin := range p.loadNewLocalPlugins(config.PluginStoragePath) {
  50. // get assets
  51. assets, err := plugin.Decoder.Assets()
  52. if err != nil {
  53. log.Error("get plugin assets error: %v", err)
  54. continue
  55. }
  56. local_plugin_runtime := local_manager.NewLocalPluginRuntime()
  57. local_plugin_runtime.PluginRuntime = plugin.Runtime
  58. local_plugin_runtime.PositivePluginRuntime = positive_manager.PositivePluginRuntime{
  59. BasicPluginRuntime: basic_manager.NewBasicPluginRuntime(p.mediaManager),
  60. LocalPackagePath: plugin.Runtime.State.AbsolutePath,
  61. WorkingPath: plugin.Runtime.State.WorkingPath,
  62. Decoder: plugin.Decoder,
  63. }
  64. if err := local_plugin_runtime.RemapAssets(
  65. &local_plugin_runtime.Config,
  66. assets,
  67. ); err != nil {
  68. log.Error("remap plugin assets error: %v", err)
  69. continue
  70. }
  71. identity, err := local_plugin_runtime.Identity()
  72. if err != nil {
  73. log.Error("get plugin identity error: %v", err)
  74. continue
  75. }
  76. // store the plugin in the storage, avoid duplicate loading
  77. p.runningPluginInStorage.Store(plugin.Runtime.State.AbsolutePath, identity.String())
  78. // local plugin
  79. routine.Submit(func() {
  80. defer func() {
  81. if r := recover(); r != nil {
  82. log.Error("plugin runtime error: %v", r)
  83. }
  84. }()
  85. // delete the plugin from the storage when the plugin is stopped
  86. defer p.runningPluginInStorage.Delete(plugin.Runtime.State.AbsolutePath)
  87. p.fullDuplexLifetime(local_plugin_runtime)
  88. })
  89. }
  90. }
  91. type pluginRuntimeWithDecoder struct {
  92. Runtime plugin_entities.PluginRuntime
  93. Decoder decoder.PluginDecoder
  94. }
  95. // chan should be closed after using that
  96. func (p *PluginManager) loadNewLocalPlugins(root_path string) <-chan *pluginRuntimeWithDecoder {
  97. ch := make(chan *pluginRuntimeWithDecoder)
  98. plugins, err := os.ReadDir(root_path)
  99. if err != nil {
  100. log.Error("no plugin found in path: %s", root_path)
  101. close(ch)
  102. return ch
  103. }
  104. routine.Submit(func() {
  105. for _, plugin := range plugins {
  106. if !plugin.IsDir() {
  107. abs_path := path.Join(root_path, plugin.Name())
  108. if _, ok := p.runningPluginInStorage.Load(abs_path); ok {
  109. // if the plugin is already running, skip it
  110. continue
  111. }
  112. plugin, err := p.loadPlugin(abs_path)
  113. if err != nil {
  114. log.Error("load plugin error: %v", err)
  115. continue
  116. }
  117. ch <- plugin
  118. }
  119. }
  120. close(ch)
  121. })
  122. return ch
  123. }
  124. func (p *PluginManager) loadPlugin(plugin_path string) (*pluginRuntimeWithDecoder, error) {
  125. pack, err := os.Open(plugin_path)
  126. if err != nil {
  127. return nil, errors.Join(err, fmt.Errorf("open plugin package error"))
  128. }
  129. defer pack.Close()
  130. if info, err := pack.Stat(); err != nil {
  131. return nil, errors.Join(err, fmt.Errorf("get plugin package info error"))
  132. } else if info.Size() > p.maxPluginPackageSize {
  133. log.Error("plugin package size is too large: %d", info.Size())
  134. return nil, err
  135. }
  136. plugin_zip, err := io.ReadAll(pack)
  137. if err != nil {
  138. return nil, errors.Join(err, fmt.Errorf("read plugin package error"))
  139. }
  140. decoder, err := decoder.NewZipPluginDecoder(plugin_zip)
  141. if err != nil {
  142. return nil, errors.Join(err, fmt.Errorf("create plugin decoder error"))
  143. }
  144. // get manifest
  145. manifest, err := decoder.Manifest()
  146. if err != nil {
  147. return nil, errors.Join(err, fmt.Errorf("get plugin manifest error"))
  148. }
  149. // check if already exists
  150. if _, exist := p.m.Load(manifest.Identity()); exist {
  151. return nil, errors.Join(fmt.Errorf("plugin already exists: %s", manifest.Identity()), err)
  152. }
  153. checksum, err := decoder.Checksum()
  154. if err != nil {
  155. return nil, errors.Join(err, fmt.Errorf("calculate checksum error"))
  156. }
  157. identity := manifest.Identity()
  158. // replace : with -
  159. identity = strings.ReplaceAll(identity, ":", "-")
  160. plugin_working_path := path.Join(p.workingDirectory, fmt.Sprintf("%s@%s", identity, checksum))
  161. // check if working directory exists
  162. if _, err := os.Stat(plugin_working_path); err == nil {
  163. return nil, errors.Join(fmt.Errorf("plugin working directory already exists: %s", plugin_working_path), err)
  164. }
  165. // extract to working directory
  166. if err := decoder.ExtractTo(plugin_working_path); err != nil {
  167. return nil, errors.Join(fmt.Errorf("extract plugin to working directory error: %v", err), err)
  168. }
  169. return &pluginRuntimeWithDecoder{
  170. Runtime: plugin_entities.PluginRuntime{
  171. Config: manifest,
  172. State: plugin_entities.PluginRuntimeState{
  173. Status: plugin_entities.PLUGIN_RUNTIME_STATUS_PENDING,
  174. Restarts: 0,
  175. AbsolutePath: plugin_path,
  176. WorkingPath: plugin_working_path,
  177. ActiveAt: nil,
  178. Verified: verifier.VerifyPlugin(decoder) == nil,
  179. },
  180. },
  181. Decoder: decoder,
  182. }, nil
  183. }