watcher.go 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244
  1. package plugin_manager
  2. import (
  3. "errors"
  4. "fmt"
  5. "io"
  6. "io/fs"
  7. "os"
  8. "path"
  9. "path/filepath"
  10. "strings"
  11. "time"
  12. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/basic_manager"
  13. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/local_manager"
  14. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/positive_manager"
  15. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/remote_manager"
  16. "github.com/langgenius/dify-plugin-daemon/internal/core/plugin_packager/decoder"
  17. "github.com/langgenius/dify-plugin-daemon/internal/types/app"
  18. "github.com/langgenius/dify-plugin-daemon/internal/types/entities/plugin_entities"
  19. "github.com/langgenius/dify-plugin-daemon/internal/utils/log"
  20. "github.com/langgenius/dify-plugin-daemon/internal/utils/routine"
  21. )
  22. func (p *PluginManager) startLocalWatcher() {
  23. go func() {
  24. // delete all plugins in working directory
  25. os.RemoveAll(p.workingDirectory)
  26. log.Info("start to handle new plugins in path: %s", p.pluginStoragePath)
  27. p.handleNewLocalPlugins()
  28. for range time.NewTicker(time.Second * 30).C {
  29. p.handleNewLocalPlugins()
  30. }
  31. }()
  32. }
  33. func (p *PluginManager) startRemoteWatcher(config *app.Config) {
  34. // launch TCP debugging server if enabled
  35. if config.PluginRemoteInstallingEnabled {
  36. server := remote_manager.NewRemotePluginServer(config, p.mediaManager)
  37. go func() {
  38. err := server.Launch()
  39. if err != nil {
  40. log.Error("start remote plugin server failed: %s", err.Error())
  41. }
  42. }()
  43. go func() {
  44. server.Wrap(func(rpr *remote_manager.RemotePluginRuntime) {
  45. identity, err := rpr.Identity()
  46. if err != nil {
  47. log.Error("get remote plugin identity failed: %s", err.Error())
  48. return
  49. }
  50. p.m.Store(identity.String(), rpr)
  51. routine.Submit(func() {
  52. defer func() {
  53. if err := recover(); err != nil {
  54. log.Error("plugin runtime error: %v", err)
  55. }
  56. p.m.Delete(identity.String())
  57. }()
  58. p.fullDuplexLifetime(rpr, nil)
  59. })
  60. })
  61. }()
  62. }
  63. }
  64. func (p *PluginManager) handleNewLocalPlugins() {
  65. // walk through all plugins
  66. err := filepath.WalkDir(p.pluginStoragePath, func(path string, d fs.DirEntry, err error) error {
  67. if err != nil {
  68. return err
  69. }
  70. if !d.IsDir() {
  71. _, _, err := p.launchLocal(path)
  72. if err != nil {
  73. log.Error("launch local plugin failed: %s", err.Error())
  74. }
  75. }
  76. return nil
  77. })
  78. if err != nil {
  79. log.Error("walk through plugins failed: %s", err.Error())
  80. }
  81. }
  82. func (p *PluginManager) launchLocal(plugin_package_path string) (
  83. plugin_entities.PluginFullDuplexLifetime, <-chan error, error,
  84. ) {
  85. plugin, err := p.getLocalPluginRuntime(plugin_package_path)
  86. if err != nil {
  87. return nil, nil, err
  88. }
  89. identity, err := plugin.decoder.UniqueIdentity()
  90. if err != nil {
  91. return nil, nil, err
  92. }
  93. // lock launch process
  94. p.localPluginLaunchingLock.Lock(identity.String())
  95. defer p.localPluginLaunchingLock.Unlock(identity.String())
  96. // check if the plugin is already running
  97. if lifetime, ok := p.m.Load(identity.String()); ok {
  98. lifetime, ok := lifetime.(plugin_entities.PluginFullDuplexLifetime)
  99. if !ok {
  100. return nil, nil, fmt.Errorf("plugin runtime not found")
  101. }
  102. // returns a closed channel to indicate the plugin is already running, no more waiting is needed
  103. c := make(chan error)
  104. close(c)
  105. return lifetime, c, nil
  106. }
  107. // extract plugin
  108. decoder, ok := plugin.decoder.(*decoder.ZipPluginDecoder)
  109. if !ok {
  110. return nil, nil, fmt.Errorf("plugin decoder is not a zip decoder")
  111. }
  112. if err := decoder.ExtractTo(plugin.runtime.State.WorkingPath); err != nil {
  113. return nil, nil, errors.Join(err, fmt.Errorf("extract plugin to working directory error"))
  114. }
  115. success := false
  116. failed := func(message string) error {
  117. if !success {
  118. os.RemoveAll(plugin.runtime.State.WorkingPath)
  119. }
  120. return errors.New(message)
  121. }
  122. // get assets
  123. assets, err := plugin.decoder.Assets()
  124. if err != nil {
  125. return nil, nil, failed(err.Error())
  126. }
  127. local_plugin_runtime := local_manager.NewLocalPluginRuntime(p.pythonInterpreterPath)
  128. local_plugin_runtime.PluginRuntime = plugin.runtime
  129. local_plugin_runtime.PositivePluginRuntime = positive_manager.PositivePluginRuntime{
  130. BasicPluginRuntime: basic_manager.NewBasicPluginRuntime(p.mediaManager),
  131. LocalPackagePath: plugin.runtime.State.AbsolutePath,
  132. WorkingPath: plugin.runtime.State.WorkingPath,
  133. Decoder: plugin.decoder,
  134. }
  135. if err := local_plugin_runtime.RemapAssets(
  136. &local_plugin_runtime.Config,
  137. assets,
  138. ); err != nil {
  139. return nil, nil, failed(errors.Join(err, fmt.Errorf("remap plugin assets error")).Error())
  140. }
  141. success = true
  142. p.m.Store(identity.String(), local_plugin_runtime)
  143. launched_chan := make(chan error)
  144. // local plugin
  145. routine.Submit(func() {
  146. defer func() {
  147. if r := recover(); r != nil {
  148. log.Error("plugin runtime panic: %v", r)
  149. }
  150. p.m.Delete(identity.String())
  151. }()
  152. p.fullDuplexLifetime(local_plugin_runtime, launched_chan)
  153. })
  154. return local_plugin_runtime, launched_chan, nil
  155. }
  156. type pluginRuntimeWithDecoder struct {
  157. runtime plugin_entities.PluginRuntime
  158. decoder decoder.PluginDecoder
  159. }
  160. // extract plugin from package to working directory
  161. func (p *PluginManager) getLocalPluginRuntime(plugin_path string) (
  162. *pluginRuntimeWithDecoder,
  163. error,
  164. ) {
  165. pack, err := os.Open(plugin_path)
  166. if err != nil {
  167. return nil, errors.Join(err, fmt.Errorf("open plugin package error"))
  168. }
  169. defer pack.Close()
  170. if info, err := pack.Stat(); err != nil {
  171. return nil, errors.Join(err, fmt.Errorf("get plugin package info error"))
  172. } else if info.Size() > p.maxPluginPackageSize {
  173. log.Error("plugin package size is too large: %d", info.Size())
  174. return nil, errors.Join(err, fmt.Errorf("plugin package size is too large"))
  175. }
  176. plugin_zip, err := io.ReadAll(pack)
  177. if err != nil {
  178. return nil, errors.Join(err, fmt.Errorf("read plugin package error"))
  179. }
  180. decoder, err := decoder.NewZipPluginDecoder(plugin_zip)
  181. if err != nil {
  182. return nil, errors.Join(err, fmt.Errorf("create plugin decoder error"))
  183. }
  184. // get manifest
  185. manifest, err := decoder.Manifest()
  186. if err != nil {
  187. return nil, errors.Join(err, fmt.Errorf("get plugin manifest error"))
  188. }
  189. checksum, err := decoder.Checksum()
  190. if err != nil {
  191. return nil, errors.Join(err, fmt.Errorf("calculate checksum error"))
  192. }
  193. identity := manifest.Identity()
  194. identity = strings.ReplaceAll(identity, ":", "-")
  195. plugin_working_path := path.Join(p.workingDirectory, fmt.Sprintf("%s@%s", identity, checksum))
  196. return &pluginRuntimeWithDecoder{
  197. runtime: plugin_entities.PluginRuntime{
  198. Config: manifest,
  199. State: plugin_entities.PluginRuntimeState{
  200. Status: plugin_entities.PLUGIN_RUNTIME_STATUS_PENDING,
  201. Restarts: 0,
  202. AbsolutePath: plugin_path,
  203. WorkingPath: plugin_working_path,
  204. ActiveAt: nil,
  205. Verified: manifest.Verified,
  206. },
  207. },
  208. decoder: decoder,
  209. }, nil
  210. }