feat: support to start extension via http (#664)

* feat: support to start extension via http

* fix the extension address checking

---------

Co-authored-by: rick <[email protected]>

Author: LinuxSuRen

cmd/server.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2023-2024 API Testing Authors.
+Copyright 2023-2025 API Testing Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -424,7 +424,7 @@ func startPlugins(storeExtMgr server.ExtManager, kinds *server.StoreKinds) (err
 	const socketPrefix = "unix://"
 
 	for _, kind := range kinds.Data {
-		if kind.Enabled && strings.HasPrefix(kind.Url, socketPrefix) {
+		if kind.Enabled && (strings.HasPrefix(kind.Url, socketPrefix) || strings.Contains(kind.Url, ":")) {
 			err = errors.Join(err, storeExtMgr.Start(kind.Name, kind.Url))
 		}
 	}

pkg/server/store_ext_manager.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2023-2024 API Testing Authors.
+Copyright 2023-2025 API Testing Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -16,160 +16,172 @@ limitations under the License.
 package server
 
 import (
-	"context"
-	"errors"
-	"fmt"
-	"io"
-	"net/http"
-	"os"
-	"path/filepath"
-	"strings"
-	sync "sync"
-	"syscall"
-	"time"
-
-	"github.com/linuxsuren/api-testing/pkg/util/home"
-
-	"github.com/linuxsuren/api-testing/pkg/downloader"
-	"github.com/linuxsuren/api-testing/pkg/logging"
-
-	fakeruntime "github.com/linuxsuren/go-fake-runtime"
+    "context"
+    "errors"
+    "fmt"
+    "io"
+    "net/http"
+    "os"
+    "path/filepath"
+    "strings"
+    sync "sync"
+    "syscall"
+    "time"
+
+    "github.com/linuxsuren/api-testing/pkg/util/home"
+
+    "github.com/linuxsuren/api-testing/pkg/downloader"
+    "github.com/linuxsuren/api-testing/pkg/logging"
+
+    fakeruntime "github.com/linuxsuren/go-fake-runtime"
 )
 
 var (
-	serverLogger = logging.DefaultLogger(logging.LogLevelInfo).WithName("server")
+    serverLogger = logging.DefaultLogger(logging.LogLevelInfo).WithName("server")
 )
 
 type ExtManager interface {
-	Start(name, socket string) (err error)
-	StopAll() (err error)
-	WithDownloader(downloader.PlatformAwareOCIDownloader)
+    Start(name, socket string) (err error)
+    StopAll() (err error)
+    WithDownloader(downloader.PlatformAwareOCIDownloader)
 }
 
 type storeExtManager struct {
-	execer               fakeruntime.Execer
-	ociDownloader        downloader.PlatformAwareOCIDownloader
-	socketPrefix         string
-	filesNeedToBeRemoved []string
-	extStatusMap         map[string]bool
-	processs             []fakeruntime.Process
-	processChan          chan fakeruntime.Process
-	stopSingal           chan struct{}
-	lock                 *sync.RWMutex
+    execer               fakeruntime.Execer
+    ociDownloader        downloader.PlatformAwareOCIDownloader
+    socketPrefix         string
+    filesNeedToBeRemoved []string
+    extStatusMap         map[string]bool
+    processs             []fakeruntime.Process
+    processChan          chan fakeruntime.Process
+    stopSingal           chan struct{}
+    lock                 *sync.RWMutex
 }
 
 var ss *storeExtManager
 
 func NewStoreExtManager(execer fakeruntime.Execer) ExtManager {
-	if ss == nil {
-		ss = &storeExtManager{
-			processChan: make(chan fakeruntime.Process),
-			stopSingal:  make(chan struct{}, 1),
-			lock:        &sync.RWMutex{},
-		}
-		ss.execer = execer
-		ss.socketPrefix = "unix://"
-		ss.extStatusMap = map[string]bool{}
-		ss.processCollect()
-		ss.WithDownloader(&nonDownloader{})
-	}
-	return ss
+    if ss == nil {
+        ss = &storeExtManager{
+            processChan: make(chan fakeruntime.Process),
+            stopSingal:  make(chan struct{}, 1),
+            lock:        &sync.RWMutex{},
+        }
+        ss.execer = execer
+        ss.socketPrefix = "unix://"
+        ss.extStatusMap = map[string]bool{}
+        ss.processCollect()
+        ss.WithDownloader(&nonDownloader{})
+    }
+    return ss
 }
 
 func NewStoreExtManagerInstance(execer fakeruntime.Execer) ExtManager {
-	ss = &storeExtManager{
-		processChan: make(chan fakeruntime.Process),
-		stopSingal:  make(chan struct{}, 1),
-		lock:        &sync.RWMutex{},
-	}
-	ss.execer = execer
-	ss.socketPrefix = "unix://"
-	ss.extStatusMap = map[string]bool{}
-	ss.processCollect()
-	ss.WithDownloader(&nonDownloader{})
-	return ss
+    ss = &storeExtManager{
+        processChan: make(chan fakeruntime.Process),
+        stopSingal:  make(chan struct{}, 1),
+        lock:        &sync.RWMutex{},
+    }
+    ss.execer = execer
+    ss.socketPrefix = "unix://"
+    ss.extStatusMap = map[string]bool{}
+    ss.processCollect()
+    ss.WithDownloader(&nonDownloader{})
+    return ss
 }
 
 func (s *storeExtManager) Start(name, socket string) (err error) {
-	if v, ok := s.extStatusMap[name]; ok && v {
-		return
-	}
-	targetDir := home.GetUserBinDir()
-	targetBinaryFile := filepath.Join(targetDir, name)
-
-	var binaryPath string
-	if _, err = os.Stat(targetBinaryFile); err == nil {
-		binaryPath = targetBinaryFile
-	} else {
-		binaryPath, err = s.execer.LookPath(name)
-		if err != nil {
-			err = fmt.Errorf("not found extension, try to download it, error: %v", err)
-			go func() {
-				reader, dErr := s.ociDownloader.Download(name, "", "")
-				if dErr != nil {
-					serverLogger.Error(dErr, "failed to download extension", "name", name)
-				} else {
-					extFile := s.ociDownloader.GetTargetFile()
-
-					targetFile := filepath.Base(extFile)
-					if dErr = downloader.WriteTo(reader, targetDir, targetFile); dErr == nil {
-						binaryPath = filepath.Join(targetDir, targetFile)
-						s.startPlugin(socket, binaryPath, name)
-					} else {
-						serverLogger.Error(dErr, "failed to save extension", "targetFile", targetFile)
-					}
-				}
-			}()
-		}
-	}
-
-	if err == nil {
-		go s.startPlugin(socket, binaryPath, name)
-	}
-	return
+    if v, ok := s.extStatusMap[name]; ok && v {
+        return
+    }
+    targetDir := home.GetUserBinDir()
+    targetBinaryFile := filepath.Join(targetDir, name)
+
+    var binaryPath string
+    if _, err = os.Stat(targetBinaryFile); err == nil {
+        binaryPath = targetBinaryFile
+    } else {
+        binaryPath, err = s.execer.LookPath(name)
+        if err != nil {
+            err = fmt.Errorf("not found extension, try to download it, error: %v", err)
+            go func() {
+                reader, dErr := s.ociDownloader.Download(name, "", "")
+                if dErr != nil {
+                    serverLogger.Error(dErr, "failed to download extension", "name", name)
+                } else {
+                    extFile := s.ociDownloader.GetTargetFile()
+
+                    targetFile := filepath.Base(extFile)
+                    if dErr = downloader.WriteTo(reader, targetDir, targetFile); dErr == nil {
+                        binaryPath = filepath.Join(targetDir, targetFile)
+                        s.startPlugin(socket, binaryPath, name)
+                    } else {
+                        serverLogger.Error(dErr, "failed to save extension", "targetFile", targetFile)
+                    }
+                }
+            }()
+        }
+    }
+
+    if err == nil {
+        go s.startPlugin(socket, binaryPath, name)
+    }
+    return
 }
 
 func (s *storeExtManager) startPlugin(socketURL, plugin, pluginName string) (err error) {
-	socketFile := strings.TrimPrefix(socketURL, s.socketPrefix)
-	_ = os.RemoveAll(socketFile) // always deleting the socket file to avoid start failing
+    if strings.Contains(socketURL, ":") && !strings.HasPrefix(socketURL, s.socketPrefix) {
+        err = s.startPluginViaHTTP(socketURL, plugin, pluginName)
+        return
+    }
+    socketFile := strings.TrimPrefix(socketURL, s.socketPrefix)
+    _ = os.RemoveAll(socketFile) // always deleting the socket file to avoid start failing
 
-	s.lock.Lock()
-	s.filesNeedToBeRemoved = append(s.filesNeedToBeRemoved, socketFile)
-	s.extStatusMap[pluginName] = true
-	s.lock.Unlock()
+    s.lock.Lock()
+    s.filesNeedToBeRemoved = append(s.filesNeedToBeRemoved, socketFile)
+    s.extStatusMap[pluginName] = true
+    s.lock.Unlock()
 
-	if err = s.execer.RunCommandWithIO(plugin, "", os.Stdout, os.Stderr, s.processChan, "--socket", socketFile); err != nil {
-		serverLogger.Info("failed to start ext manager", "socket", socketURL, "error: ", err.Error())
-	}
-	return
+    if err = s.execer.RunCommandWithIO(plugin, "", os.Stdout, os.Stderr, s.processChan, "--socket", socketFile); err != nil {
+        serverLogger.Info("failed to start ext manager", "socket", socketURL, "error: ", err.Error())
+    }
+    return
+}
+
+func (s *storeExtManager) startPluginViaHTTP(httpURL, plugin, pluginName string) (err error) {
+    port := strings.Split(httpURL, ":")[1]
+    if err = s.execer.RunCommandWithIO(plugin, "", os.Stdout, os.Stderr, s.processChan, "--port", port); err != nil {
+        serverLogger.Info("failed to start ext manager", "port", port, "error: ", err.Error())
+    }
+    return
 }
 
 func (s *storeExtManager) StopAll() error {
-	serverLogger.Info("stop", "extensions", len(s.processs))
-	for _, p := range s.processs {
-		if p != nil {
-			p.Signal(syscall.SIGTERM)
-		}
-	}
-	s.stopSingal <- struct{}{}
-	return nil
+    serverLogger.Info("stop", "extensions", len(s.processs))
+    for _, p := range s.processs {
+        if p != nil {
+            p.Signal(syscall.SIGTERM)
+        }
+    }
+    s.stopSingal <- struct{}{}
+    return nil
 }
 
 func (s *storeExtManager) WithDownloader(ociDownloader downloader.PlatformAwareOCIDownloader) {
-	s.ociDownloader = ociDownloader
+    s.ociDownloader = ociDownloader
 }
 
 func (s *storeExtManager) processCollect() {
-	go func() {
-		for {
-			select {
-			case p := <-s.processChan:
-				s.processs = append(s.processs, p)
-			case <-s.stopSingal:
-				return
-			}
-		}
-	}()
+    go func() {
+        for {
+            select {
+            case p := <-s.processChan:
+                s.processs = append(s.processs, p)
+            case <-s.stopSingal:
+                return
+            }
+        }
+    }()
 }
 
 var ErrDownloadNotSupport = errors.New("no support")
@@ -179,44 +191,44 @@ type nonDownloader struct{}
 var _ downloader.PlatformAwareOCIDownloader = &nonDownloader{}
 
 func (n *nonDownloader) WithBasicAuth(username string, password string) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (n *nonDownloader) Download(image, tag, file string) (reader io.Reader, err error) {
-	err = ErrDownloadNotSupport
-	return
+    err = ErrDownloadNotSupport
+    return
 }
 
 func (n *nonDownloader) WithOS(string) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (n *nonDownloader) WithArch(string) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (n *nonDownloader) WithRegistry(string) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (n *nonDownloader) WithKind(string) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (n *nonDownloader) WithImagePrefix(imagePrefix string) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 func (d *nonDownloader) WithRoundTripper(rt http.RoundTripper) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (d *nonDownloader) WithInsecure(bool) {
-	// Do nothing because this is an empty implementation
+    // Do nothing because this is an empty implementation
 }
 
 func (d *nonDownloader) WithTimeout(time.Duration)   {}
 func (d *nonDownloader) WithContext(context.Context) {}
 
 func (n *nonDownloader) GetTargetFile() string {
-	return ""
+    return ""
 }