refactor: move remaining api packages to core (#1731)

* core 1

* api/openai/files fix

* core 2 - core/config

* move over core api.go and tests to the start of core/http

* move over localai specific endpoints to core/http, begin the service/endpoint split there

* refactor big chunk on the plane

* refactor chunk 2 on plane, next step: port and modify changes to request.go

* easy fixes for request.go, major changes not done yet

* lintfix

* json tag lintfix?

* gitignore and .keep files

* strange fix attempt: rename the config dir?

core/services/backend_monitor.go

@@ -0,0 +1,140 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/go-skynet/LocalAI/core/config"
+	"github.com/go-skynet/LocalAI/core/schema"
+	"github.com/go-skynet/LocalAI/pkg/grpc/proto"
+	"github.com/go-skynet/LocalAI/pkg/model"
+
+	"github.com/rs/zerolog/log"
+
+	gopsutil "github.com/shirou/gopsutil/v3/process"
+)
+
+type BackendMonitor struct {
+	configLoader *config.BackendConfigLoader
+	modelLoader  *model.ModelLoader
+	options      *config.ApplicationConfig // Taking options in case we need to inspect ExternalGRPCBackends, though that's out of scope for now, hence the name.
+}
+
+func NewBackendMonitor(configLoader *config.BackendConfigLoader, modelLoader *model.ModelLoader, appConfig *config.ApplicationConfig) BackendMonitor {
+	return BackendMonitor{
+		configLoader: configLoader,
+		modelLoader:  modelLoader,
+		options:      appConfig,
+	}
+}
+
+func (bm BackendMonitor) getModelLoaderIDFromModelName(modelName string) (string, error) {
+	config, exists := bm.configLoader.GetBackendConfig(modelName)
+	var backendId string
+	if exists {
+		backendId = config.Model
+	} else {
+		// Last ditch effort: use it raw, see if a backend happens to match.
+		backendId = modelName
+	}
+
+	if !strings.HasSuffix(backendId, ".bin") {
+		backendId = fmt.Sprintf("%s.bin", backendId)
+	}
+
+	return backendId, nil
+}
+
+func (bm *BackendMonitor) SampleLocalBackendProcess(model string) (*schema.BackendMonitorResponse, error) {
+	config, exists := bm.configLoader.GetBackendConfig(model)
+	var backend string
+	if exists {
+		backend = config.Model
+	} else {
+		// Last ditch effort: use it raw, see if a backend happens to match.
+		backend = model
+	}
+
+	if !strings.HasSuffix(backend, ".bin") {
+		backend = fmt.Sprintf("%s.bin", backend)
+	}
+
+	pid, err := bm.modelLoader.GetGRPCPID(backend)
+
+	if err != nil {
+		log.Error().Msgf("model %s : failed to find pid %+v", model, err)
+		return nil, err
+	}
+
+	// Name is slightly frightening but this does _not_ create a new process, rather it looks up an existing process by PID.
+	backendProcess, err := gopsutil.NewProcess(int32(pid))
+
+	if err != nil {
+		log.Error().Msgf("model %s [PID %d] : error getting process info %+v", model, pid, err)
+		return nil, err
+	}
+
+	memInfo, err := backendProcess.MemoryInfo()
+
+	if err != nil {
+		log.Error().Msgf("model %s [PID %d] : error getting memory info %+v", model, pid, err)
+		return nil, err
+	}
+
+	memPercent, err := backendProcess.MemoryPercent()
+	if err != nil {
+		log.Error().Msgf("model %s [PID %d] : error getting memory percent %+v", model, pid, err)
+		return nil, err
+	}
+
+	cpuPercent, err := backendProcess.CPUPercent()
+	if err != nil {
+		log.Error().Msgf("model %s [PID %d] : error getting cpu percent %+v", model, pid, err)
+		return nil, err
+	}
+
+	return &schema.BackendMonitorResponse{
+		MemoryInfo:    memInfo,
+		MemoryPercent: memPercent,
+		CPUPercent:    cpuPercent,
+	}, nil
+}
+
+func (bm BackendMonitor) CheckAndSample(modelName string) (*proto.StatusResponse, error) {
+	backendId, err := bm.getModelLoaderIDFromModelName(modelName)
+	if err != nil {
+		return nil, err
+	}
+	modelAddr := bm.modelLoader.CheckIsLoaded(backendId)
+	if modelAddr == "" {
+		return nil, fmt.Errorf("backend %s is not currently loaded", backendId)
+	}
+
+	status, rpcErr := modelAddr.GRPC(false, nil).Status(context.TODO())
+	if rpcErr != nil {
+		log.Warn().Msgf("backend %s experienced an error retrieving status info: %s", backendId, rpcErr.Error())
+		val, slbErr := bm.SampleLocalBackendProcess(backendId)
+		if slbErr != nil {
+			return nil, fmt.Errorf("backend %s experienced an error retrieving status info via rpc: %s, then failed local node process sample: %s", backendId, rpcErr.Error(), slbErr.Error())
+		}
+		return &proto.StatusResponse{
+			State: proto.StatusResponse_ERROR,
+			Memory: &proto.MemoryUsageData{
+				Total: val.MemoryInfo.VMS,
+				Breakdown: map[string]uint64{
+					"gopsutil-RSS": val.MemoryInfo.RSS,
+				},
+			},
+		}, nil
+	}
+	return status, nil
+}
+
+func (bm BackendMonitor) ShutdownModel(modelName string) error {
+	backendId, err := bm.getModelLoaderIDFromModelName(modelName)
+	if err != nil {
+		return err
+	}
+	return bm.modelLoader.ShutdownModel(backendId)
+}

core/services/gallery.go

@@ -0,0 +1,167 @@
+package services
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/go-skynet/LocalAI/core/config"
+	"github.com/go-skynet/LocalAI/pkg/gallery"
+	"github.com/go-skynet/LocalAI/pkg/utils"
+	"gopkg.in/yaml.v2"
+)
+
+type GalleryService struct {
+	modelPath string
+	sync.Mutex
+	C        chan gallery.GalleryOp
+	statuses map[string]*gallery.GalleryOpStatus
+}
+
+func NewGalleryService(modelPath string) *GalleryService {
+	return &GalleryService{
+		modelPath: modelPath,
+		C:         make(chan gallery.GalleryOp),
+		statuses:  make(map[string]*gallery.GalleryOpStatus),
+	}
+}
+
+func prepareModel(modelPath string, req gallery.GalleryModel, cl *config.BackendConfigLoader, downloadStatus func(string, string, string, float64)) error {
+
+	config, err := gallery.GetGalleryConfigFromURL(req.URL)
+	if err != nil {
+		return err
+	}
+
+	config.Files = append(config.Files, req.AdditionalFiles...)
+
+	return gallery.InstallModel(modelPath, req.Name, &config, req.Overrides, downloadStatus)
+}
+
+func (g *GalleryService) UpdateStatus(s string, op *gallery.GalleryOpStatus) {
+	g.Lock()
+	defer g.Unlock()
+	g.statuses[s] = op
+}
+
+func (g *GalleryService) GetStatus(s string) *gallery.GalleryOpStatus {
+	g.Lock()
+	defer g.Unlock()
+
+	return g.statuses[s]
+}
+
+func (g *GalleryService) GetAllStatus() map[string]*gallery.GalleryOpStatus {
+	g.Lock()
+	defer g.Unlock()
+
+	return g.statuses
+}
+
+func (g *GalleryService) Start(c context.Context, cl *config.BackendConfigLoader) {
+	go func() {
+		for {
+			select {
+			case <-c.Done():
+				return
+			case op := <-g.C:
+				utils.ResetDownloadTimers()
+
+				g.UpdateStatus(op.Id, &gallery.GalleryOpStatus{Message: "processing", Progress: 0})
+
+				// updates the status with an error
+				updateError := func(e error) {
+					g.UpdateStatus(op.Id, &gallery.GalleryOpStatus{Error: e, Processed: true, Message: "error: " + e.Error()})
+				}
+
+				// displayDownload displays the download progress
+				progressCallback := func(fileName string, current string, total string, percentage float64) {
+					g.UpdateStatus(op.Id, &gallery.GalleryOpStatus{Message: "processing", FileName: fileName, Progress: percentage, TotalFileSize: total, DownloadedFileSize: current})
+					utils.DisplayDownloadFunction(fileName, current, total, percentage)
+				}
+
+				var err error
+				// if the request contains a gallery name, we apply the gallery from the gallery list
+				if op.GalleryName != "" {
+					if strings.Contains(op.GalleryName, "@") {
+						err = gallery.InstallModelFromGallery(op.Galleries, op.GalleryName, g.modelPath, op.Req, progressCallback)
+					} else {
+						err = gallery.InstallModelFromGalleryByName(op.Galleries, op.GalleryName, g.modelPath, op.Req, progressCallback)
+					}
+				} else {
+					err = prepareModel(g.modelPath, op.Req, cl, progressCallback)
+				}
+
+				if err != nil {
+					updateError(err)
+					continue
+				}
+
+				// Reload models
+				err = cl.LoadBackendConfigsFromPath(g.modelPath)
+				if err != nil {
+					updateError(err)
+					continue
+				}
+
+				err = cl.Preload(g.modelPath)
+				if err != nil {
+					updateError(err)
+					continue
+				}
+
+				g.UpdateStatus(op.Id, &gallery.GalleryOpStatus{Processed: true, Message: "completed", Progress: 100})
+			}
+		}
+	}()
+}
+
+type galleryModel struct {
+	gallery.GalleryModel `yaml:",inline"` // https://github.com/go-yaml/yaml/issues/63
+	ID                   string           `json:"id"`
+}
+
+func processRequests(modelPath, s string, cm *config.BackendConfigLoader, galleries []gallery.Gallery, requests []galleryModel) error {
+	var err error
+	for _, r := range requests {
+		utils.ResetDownloadTimers()
+		if r.ID == "" {
+			err = prepareModel(modelPath, r.GalleryModel, cm, utils.DisplayDownloadFunction)
+		} else {
+			if strings.Contains(r.ID, "@") {
+				err = gallery.InstallModelFromGallery(
+					galleries, r.ID, modelPath, r.GalleryModel, utils.DisplayDownloadFunction)
+			} else {
+				err = gallery.InstallModelFromGalleryByName(
+					galleries, r.ID, modelPath, r.GalleryModel, utils.DisplayDownloadFunction)
+			}
+		}
+	}
+	return err
+}
+
+func ApplyGalleryFromFile(modelPath, s string, cl *config.BackendConfigLoader, galleries []gallery.Gallery) error {
+	dat, err := os.ReadFile(s)
+	if err != nil {
+		return err
+	}
+	var requests []galleryModel
+
+	if err := yaml.Unmarshal(dat, &requests); err != nil {
+		return err
+	}
+
+	return processRequests(modelPath, s, cl, galleries, requests)
+}
+
+func ApplyGalleryFromString(modelPath, s string, cl *config.BackendConfigLoader, galleries []gallery.Gallery) error {
+	var requests []galleryModel
+	err := json.Unmarshal([]byte(s), &requests)
+	if err != nil {
+		return err
+	}
+
+	return processRequests(modelPath, s, cl, galleries, requests)
+}

core/services/metrics.go

@@ -0,0 +1,54 @@
+package services
+
+import (
+	"context"
+
+	"github.com/rs/zerolog/log"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/prometheus"
+	"go.opentelemetry.io/otel/metric"
+	metricApi "go.opentelemetry.io/otel/sdk/metric"
+)
+
+type LocalAIMetricsService struct {
+	Meter         metric.Meter
+	ApiTimeMetric metric.Float64Histogram
+}
+
+func (m *LocalAIMetricsService) ObserveAPICall(method string, path string, duration float64) {
+	opts := metric.WithAttributes(
+		attribute.String("method", method),
+		attribute.String("path", path),
+	)
+	m.ApiTimeMetric.Record(context.Background(), duration, opts)
+}
+
+// setupOTelSDK bootstraps the OpenTelemetry pipeline.
+// If it does not return an error, make sure to call shutdown for proper cleanup.
+func NewLocalAIMetricsService() (*LocalAIMetricsService, error) {
+	exporter, err := prometheus.New()
+	if err != nil {
+		return nil, err
+	}
+	provider := metricApi.NewMeterProvider(metricApi.WithReader(exporter))
+	meter := provider.Meter("github.com/go-skynet/LocalAI")
+
+	apiTimeMetric, err := meter.Float64Histogram("api_call", metric.WithDescription("api calls"))
+	if err != nil {
+		return nil, err
+	}
+
+	return &LocalAIMetricsService{
+		Meter:         meter,
+		ApiTimeMetric: apiTimeMetric,
+	}, nil
+}
+
+func (lams LocalAIMetricsService) Shutdown() error {
+	// TODO: Not sure how to actually do this:
+	//// setupOTelSDK bootstraps the OpenTelemetry pipeline.
+	//// If it does not return an error, make sure to call shutdown for proper cleanup.
+
+	log.Warn().Msgf("LocalAIMetricsService Shutdown called, but OTelSDK proper shutdown not yet implemented?")
+	return nil
+}