dearwolf/LocalAI
1
0
Fork 0
mirror of https://github.com/mudler/LocalAI.git synced 2025-05-29 23:15:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 15

Rewrite P2P cmd flags handling; add peerguard/auth edgevpn support; WIP edgevpn ledger management http API

Browse source
This commit is contained in:
mintyleaf 2025-03-29 06:45:00 +04:00
parent c965197d6f
commit aa7171dd5d
14 changed files with 306 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

13
core/cli/api/p2p.go
View file

@ -8,12 +8,13 @@ import (
"strings"
"github.com/mudler/LocalAI/core/p2p"
p2pConfig "github.com/mudler/edgevpn/pkg/config"
"github.com/mudler/edgevpn/pkg/node"
"github.com/rs/zerolog/log"
)
func StartP2PStack(ctx context.Context, address, token, networkID string, federated bool) error {
func StartP2PStack(ctx context.Context, p2pCfg p2pConfig.Config, address, networkID string, federated bool) error {
var n *node.Node
// Here we are avoiding creating multiple nodes:
// - if the federated mode is enabled, we create a federated node and expose a service
@ -29,12 +30,12 @@ func StartP2PStack(ctx context.Context, address, token, networkID string, federa
// Here a new node is created and started
// and a service is exposed by the node
node, err := p2p.ExposeService(ctx, "localhost", port, token, p2p.NetworkID(networkID, p2p.FederatedID))
node, err := p2p.ExposeService(ctx, p2pCfg, "localhost", port, p2p.NetworkID(networkID, p2p.FederatedID))
if err != nil {
return err
}
if err := p2p.ServiceDiscoverer(ctx, node, token, p2p.NetworkID(networkID, p2p.FederatedID), nil, false); err != nil {
if err := p2p.ServiceDiscoverer(ctx, node, p2p.NetworkID(networkID, p2p.FederatedID), nil, false); err != nil {
return err
}
@ -42,10 +43,10 @@ func StartP2PStack(ctx context.Context, address, token, networkID string, federa
}
// If the p2p mode is enabled, we start the service discovery
if token != "" {
if p2pCfg.NetworkToken != "" {
// If a node wasn't created previously, create it
if n == nil {
node, err := p2p.NewNode(token)
node, err := p2p.NewNode(p2pCfg)
if err != nil {
return err
}
@ -58,7 +59,7 @@ func StartP2PStack(ctx context.Context, address, token, networkID string, federa
// Attach a ServiceDiscoverer to the p2p node
log.Info().Msg("Starting P2P server discovery...")
if err := p2p.ServiceDiscoverer(ctx, n, token, p2p.NetworkID(networkID, p2p.WorkerID), func(serviceID string, node p2p.NodeData) {
if err := p2p.ServiceDiscoverer(ctx, n, p2p.NetworkID(networkID, p2p.WorkerID), func(serviceID string, node p2p.NodeData) {
var tunnelAddresses []string
for _, v := range p2p.GetAvailableNodes(p2p.NetworkID(networkID, p2p.WorkerID)) {
if v.IsOnline() {

7
core/cli/explorer.go
View file

@ -5,11 +5,14 @@ import (
"time"
cliContext "github.com/mudler/LocalAI/core/cli/context"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/LocalAI/core/explorer"
"github.com/mudler/LocalAI/core/http"
)
type ExplorerCMD struct {
cliP2P.P2PCommonFlags `embed:""`
Address string `env:"LOCALAI_ADDRESS,ADDRESS" default:":8080" help:"Bind address for the API server" group:"api"`
PoolDatabase string `env:"LOCALAI_POOL_DATABASE,POOL_DATABASE" default:"explorer.json" help:"Path to the pool database" group:"api"`
ConnectionTimeout string `env:"LOCALAI_CONNECTION_TIMEOUT,CONNECTION_TIMEOUT" default:"2m" help:"Connection timeout for the explorer" group:"api"`
@ -33,14 +36,14 @@ func (e *ExplorerCMD) Run(ctx *cliContext.Context) error {
if e.WithSync {
ds := explorer.NewDiscoveryServer(db, dur, e.ConnectionErrorThreshold)
go ds.Start(context.Background(), true)
go ds.Start(context.Background(), e.P2PCommonFlags, true)
}
if e.OnlySync {
ds := explorer.NewDiscoveryServer(db, dur, e.ConnectionErrorThreshold)
ctx := context.Background()
return ds.Start(ctx, false)
return ds.Start(ctx, e.P2PCommonFlags, false)
}
appHTTP := http.Explorer(db)

13
core/cli/federated.go
View file

@ -4,20 +4,21 @@ import (
"context"
cliContext "github.com/mudler/LocalAI/core/cli/context"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/LocalAI/core/p2p"
)
type FederatedCLI struct {
Address string `env:"LOCALAI_ADDRESS,ADDRESS" default:":8080" help:"Bind address for the API server" group:"api"`
Peer2PeerToken string `env:"LOCALAI_P2P_TOKEN,P2P_TOKEN,TOKEN" name:"p2ptoken" help:"Token for P2P mode (optional)" group:"p2p"`
RandomWorker bool `env:"LOCALAI_RANDOM_WORKER,RANDOM_WORKER" default:"false" help:"Select a random worker from the pool" group:"p2p"`
Peer2PeerNetworkID string `env:"LOCALAI_P2P_NETWORK_ID,P2P_NETWORK_ID" help:"Network ID for P2P mode, can be set arbitrarly by the user for grouping a set of instances." group:"p2p"`
TargetWorker string `env:"LOCALAI_TARGET_WORKER,TARGET_WORKER" help:"Target worker to run the federated server on" group:"p2p"`
cliP2P.P2PCommonFlags `embed:""`
Address string `env:"LOCALAI_ADDRESS,ADDRESS" default:":8080" help:"Bind address for the API server" group:"api"`
RandomWorker bool `env:"LOCALAI_RANDOM_WORKER,RANDOM_WORKER" default:"false" help:"Select a random worker from the pool" group:"p2p"`
TargetWorker string `env:"LOCALAI_TARGET_WORKER,TARGET_WORKER" help:"Target worker to run the federated server on" group:"p2p"`
}
func (f *FederatedCLI) Run(ctx *cliContext.Context) error {
fs := p2p.NewFederatedServer(f.Address, p2p.NetworkID(f.Peer2PeerNetworkID, p2p.FederatedID), f.Peer2PeerToken, !f.RandomWorker, f.TargetWorker)
return fs.Start(context.Background())
return fs.Start(context.Background(), f.P2PCommonFlags)
}

17
core/cli/p2p/p2p.go Normal file
View file

@ -0,0 +1,17 @@
package cli
type P2PCommonFlags struct {
Peer2PeerNoDHT bool `env:"LOCALAI_P2P_DISABLE_DHT,P2P_DISABLE_DHT" name:"p2p-disable-dht" help:"Disable DHT" group:"p2p"`
Peer2PeerLimit bool `env:"LOCALAI_P2P_ENABLE_LIMITS,P2P_ENABLE_LIMITS" name:"p2p-enable-limits" help:"Enable Limits" group:"p2p"`
Peer2PeerListenAddrs []string `env:"LOCALAI_P2P_LISTEN_MADDRS,P2P_LISTEN_MADDRS" name:"p2p-listen-maddrs" help:"A list of listen multiaddresses" group:"p2p"`
Peer2PeerBootAddrs []string `env:"LOCALAI_P2P_BOOTSTRAP_PEERS_MADDRS,P2P_BOOTSTRAP_PEERS_MADDRS" name:"p2p-bootstrap-peers-maddrs" help:"A list of bootstrap peers multiaddresses" group:"p2p"`
Peer2PeerDHTAnnounceAddrs []string `env:"LOCALAI_P2P_DHT_ANNOUNCE_MADDRS,P2P_DHT_ANNOUNCE_MADDRS" name:"p2p-dht-announce-maddrs" help:"A list of DHT announce maddrs" group:"p2p"`
Peer2PeerLibLoglevel string `env:"LOCALAI_P2P_LIB_LOGLEVEL,P2P_LIB_LOGLEVEL" name:"p2p-lib-loglevel" help:"libp2p specific loglevel" group:"p2p"`
Peer2PeerDHTInterval int `env:"LOCALAI_P2P_DHT_INTERVAL,P2P_DHT_INTERVAL" default:"360" name:"p2p-dht-interval" help:"Interval for DHT refresh (used during token generation)" group:"p2p"`
Peer2PeerOTPInterval int `env:"LOCALAI_P2P_OTP_INTERVAL,P2P_OTP_INTERVAL" default:"9000" name:"p2p-otp-interval" help:"Interval for OTP refresh (used during token generation)" group:"p2p"`
Peer2PeerToken string `env:"LOCALAI_P2P_TOKEN,P2P_TOKEN,TOKEN" name:"p2ptoken" help:"Token for P2P mode (optional)" group:"p2p"`
Peer2PeerPrivkey string `env:"LOCALAI_P2P_PRIVKEY,P2P_PRIVKEY" name:"p2pprivkey" help:"A base64 encoded protobuf serialized private key used for fixed ID (edgevpn can be used for generating one)" group:"p2p"`
Peer2PeerUsePeerguard bool `env:"LOCALAI_P2P_PEERGUARD,P2P_PEERGUARD" name:"p2ppeerguard" help:"Enable peerguarding through ecdsa authorization of nodes" group:"p2p"`
Peer2PeerAuthProvders string `env:"LOCALAI_P2P_PEERGATE_AUTH,P2P_PEERGATE_AUTH" name:"p2pauth" help:"JSON dict string with '{authProviderName: {providerOpt: value}}' structure, see edgevpn project" group:"p2p"`
Peer2PeerNetworkID string `env:"LOCALAI_P2P_NETWORK_ID,P2P_NETWORK_ID" help:"Network ID for P2P mode, can be set arbitrarly by the user for grouping a set of instances" group:"p2p"`
}

27
core/cli/run.go
View file

@ -9,6 +9,7 @@ import (
"github.com/mudler/LocalAI/core/application"
cli_api "github.com/mudler/LocalAI/core/cli/api"
cliContext "github.com/mudler/LocalAI/core/cli/context"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/LocalAI/core/config"
"github.com/mudler/LocalAI/core/http"
"github.com/mudler/LocalAI/core/p2p"
@ -55,10 +56,7 @@ type RunCMD struct {
DisableMetricsEndpoint bool `env:"LOCALAI_DISABLE_METRICS_ENDPOINT,DISABLE_METRICS_ENDPOINT" default:"false" help:"Disable the /metrics endpoint" group:"api"`
HttpGetExemptedEndpoints []string `env:"LOCALAI_HTTP_GET_EXEMPTED_ENDPOINTS" default:"^/$,^/browse/?$,^/talk/?$,^/p2p/?$,^/chat/?$,^/text2image/?$,^/tts/?$,^/static/.*$,^/swagger.*$" help:"If LOCALAI_DISABLE_API_KEY_REQUIREMENT_FOR_HTTP_GET is overriden to true, this is the list of endpoints to exempt. Only adjust this in case of a security incident or as a result of a personal security posture review" group:"hardening"`
Peer2Peer bool `env:"LOCALAI_P2P,P2P" name:"p2p" default:"false" help:"Enable P2P mode" group:"p2p"`
Peer2PeerDHTInterval int `env:"LOCALAI_P2P_DHT_INTERVAL,P2P_DHT_INTERVAL" default:"360" name:"p2p-dht-interval" help:"Interval for DHT refresh (used during token generation)" group:"p2p"`
Peer2PeerOTPInterval int `env:"LOCALAI_P2P_OTP_INTERVAL,P2P_OTP_INTERVAL" default:"9000" name:"p2p-otp-interval" help:"Interval for OTP refresh (used during token generation)" group:"p2p"`
Peer2PeerToken string `env:"LOCALAI_P2P_TOKEN,P2P_TOKEN,TOKEN" name:"p2ptoken" help:"Token for P2P mode (optional)" group:"p2p"`
Peer2PeerNetworkID string `env:"LOCALAI_P2P_NETWORK_ID,P2P_NETWORK_ID" help:"Network ID for P2P mode, can be set arbitrarly by the user for grouping a set of instances" group:"p2p"`
cliP2P.P2PCommonFlags `embed:""`
ParallelRequests bool `env:"LOCALAI_PARALLEL_REQUESTS,PARALLEL_REQUESTS" help:"Enable backends to handle multiple requests in parallel if they support it (e.g.: llama.cpp or vllm)" group:"backends"`
SingleActiveBackend bool `env:"LOCALAI_SINGLE_ACTIVE_BACKEND,SINGLE_ACTIVE_BACKEND" help:"Allow only one backend to be run at a time" group:"backends"`
PreloadBackendOnly bool `env:"LOCALAI_PRELOAD_BACKEND_ONLY,PRELOAD_BACKEND_ONLY" default:"false" help:"Do not launch the API services, only the preloaded models / backends are started (useful for multi-node setups)" group:"backends"`
@ -114,14 +112,24 @@ func (r *RunCMD) Run(ctx *cliContext.Context) error {
}
token := ""
p2pCfg := p2p.NewP2PConfig(r.P2PCommonFlags)
if r.Peer2Peer || r.Peer2PeerToken != "" {
log.Info().Msg("P2P mode enabled")
token = r.Peer2PeerToken
if token == "" {
// IF no token is provided, and p2p is enabled,
// we generate one and wait for the user to pick up the token (this is for interactive)
log.Info().Msg("No token provided, generating one")
token = p2p.GenerateToken(r.Peer2PeerDHTInterval, r.Peer2PeerOTPInterval)
connectionData, err := p2p.GenerateNewConnectionData(
r.Peer2PeerDHTInterval, r.Peer2PeerOTPInterval,
r.Peer2PeerPrivkey, r.Peer2PeerUsePeerguard,
)
if err != nil {
log.Warn().Msgf("Error generating token: %s", err.Error())
}
token = connectionData.Base64()
log.Info().Msg("Generated Token:")
fmt.Println(token)
@ -129,11 +137,18 @@ func (r *RunCMD) Run(ctx *cliContext.Context) error {
fmt.Printf("export TOKEN=\"%s\"\nlocal-ai worker p2p-llama-cpp-rpc\n", token)
}
opts = append(opts, config.WithP2PToken(token))
if r.Federated {
p2pCfg.PeerGuard.Autocleanup = true
p2pCfg.PeerGuard.PeerGate = true
}
p2pCfg.NetworkToken = token
}
backgroundCtx := context.Background()
if err := cli_api.StartP2PStack(backgroundCtx, r.Address, token, r.Peer2PeerNetworkID, r.Federated); err != nil {
if err := cli_api.StartP2PStack(backgroundCtx, p2pCfg, r.Address, r.Peer2PeerNetworkID, r.Federated); err != nil {
return err
}

20
core/cli/worker/worker_p2p.go
View file

@ -12,6 +12,7 @@ import (
"time"
cliContext "github.com/mudler/LocalAI/core/cli/context"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/LocalAI/core/p2p"
"github.com/mudler/LocalAI/pkg/assets"
"github.com/mudler/LocalAI/pkg/library"
@ -20,12 +21,13 @@ import (
)
type P2P struct {
WorkerFlags `embed:""`
Token string `env:"LOCALAI_TOKEN,LOCALAI_P2P_TOKEN,TOKEN" help:"P2P token to use"`
NoRunner bool `env:"LOCALAI_NO_RUNNER,NO_RUNNER" help:"Do not start the llama-cpp-rpc-server"`
RunnerAddress string `env:"LOCALAI_RUNNER_ADDRESS,RUNNER_ADDRESS" help:"Address of the llama-cpp-rpc-server"`
RunnerPort string `env:"LOCALAI_RUNNER_PORT,RUNNER_PORT" help:"Port of the llama-cpp-rpc-server"`
Peer2PeerNetworkID string `env:"LOCALAI_P2P_NETWORK_ID,P2P_NETWORK_ID" help:"Network ID for P2P mode, can be set arbitrarly by the user for grouping a set of instances" group:"p2p"`
WorkerFlags `embed:""`
cliP2P.P2PCommonFlags `embed:""`
Token string `env:"LOCALAI_TOKEN,LOCALAI_P2P_TOKEN,TOKEN" help:"P2P token to use"`
NoRunner bool `env:"LOCALAI_NO_RUNNER,NO_RUNNER" help:"Do not start the llama-cpp-rpc-server"`
RunnerAddress string `env:"LOCALAI_RUNNER_ADDRESS,RUNNER_ADDRESS" help:"Address of the llama-cpp-rpc-server"`
RunnerPort string `env:"LOCALAI_RUNNER_PORT,RUNNER_PORT" help:"Port of the llama-cpp-rpc-server"`
}
func (r *P2P) Run(ctx *cliContext.Context) error {
@ -41,6 +43,8 @@ func (r *P2P) Run(ctx *cliContext.Context) error {
if r.Token == "" {
return fmt.Errorf("Token is required")
}
p2pCfg := p2p.NewP2PConfig(r.P2PCommonFlags)
p2pCfg.NetworkToken = r.Token
port, err := freeport.GetFreePort()
if err != nil {
@ -60,7 +64,7 @@ func (r *P2P) Run(ctx *cliContext.Context) error {
p = r.RunnerPort
}
_, err = p2p.ExposeService(context.Background(), address, p, r.Token, p2p.NetworkID(r.Peer2PeerNetworkID, p2p.WorkerID))
_, err = p2p.ExposeService(context.Background(), p2pCfg, address, p, p2p.NetworkID(r.Peer2PeerNetworkID, p2p.WorkerID))
if err != nil {
return err
}
@ -103,7 +107,7 @@ func (r *P2P) Run(ctx *cliContext.Context) error {
}
}()
_, err = p2p.ExposeService(context.Background(), address, fmt.Sprint(port), r.Token, p2p.NetworkID(r.Peer2PeerNetworkID, p2p.WorkerID))
_, err = p2p.ExposeService(context.Background(), p2pCfg, address, fmt.Sprint(port), p2p.NetworkID(r.Peer2PeerNetworkID, p2p.WorkerID))
if err != nil {
return err
}

12
core/explorer/discovery.go
View file

@ -9,6 +9,7 @@ import (
"github.com/rs/zerolog/log"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/LocalAI/core/p2p"
"github.com/mudler/edgevpn/pkg/blockchain"
)
@ -40,7 +41,7 @@ type Network struct {
Clusters []ClusterData
}
func (s *DiscoveryServer) runBackground() {
func (s *DiscoveryServer) runBackground(p2pCommonFlags cliP2P.P2PCommonFlags) {
if len(s.database.TokenList()) == 0 {
time.Sleep(5 * time.Second) // avoid busy loop
return
@ -50,11 +51,14 @@ func (s *DiscoveryServer) runBackground() {
c, cancel := context.WithTimeout(context.Background(), s.connectionTime)
defer cancel()
p2pCfg := p2p.NewP2PConfig(p2pCommonFlags)
p2pCfg.NetworkToken = token
// Connect to the network
// Get the number of nodes
// save it in the current state (mutex)
// do not do in parallel
n, err := p2p.NewNode(token)
n, err := p2p.NewNode(p2pCfg)
if err != nil {
log.Err(err).Msg("Failed to create node")
s.failedToken(token)
@ -197,14 +201,14 @@ func (s *DiscoveryServer) retrieveNetworkData(c context.Context, ledger *blockch
}
// Start the discovery server. This is meant to be run in to a goroutine.
func (s *DiscoveryServer) Start(ctx context.Context, keepRunning bool) error {
func (s *DiscoveryServer) Start(ctx context.Context, p2pCommonFlags cliP2P.P2PCommonFlags, keepRunning bool) error {
for {
select {
case <-ctx.Done():
return fmt.Errorf("context cancelled")
default:
// Collect data
s.runBackground()
s.runBackground(p2pCommonFlags)
if !keepRunning {
return nil
}

80
core/http/routes/peerguard.go Normal file
View file

@ -0,0 +1,80 @@
package routes
import (
"context"
"time"
"github.com/gofiber/fiber/v2"
"github.com/mudler/edgevpn/pkg/node"
)
const DefaultInterval = 5 * time.Second
const Timeout = 20 * time.Second
// TODO connect routes and write a middleware for authorization based on p2p auth providers private keys
func RegisterPeerguardAuthRoutes(app *fiber.App, e *node.Node) {
app.Get("ledger/:bucket/:key", func(c *fiber.Ctx) error {
bucket := c.Params("bucket")
key := c.Params("key")
ledger, err := e.Ledger()
if err != nil {
return err
}
return c.JSON(ledger.CurrentData()[bucket][key])
})
app.Get("ledger/:bucket", func(c *fiber.Ctx) error {
bucket := c.Params("bucket")
ledger, err := e.Ledger()
if err != nil {
return err
}
return c.JSON(ledger.CurrentData()[bucket])
})
announcing := struct{ State string }{"Announcing"}
// Store arbitrary data
app.Get("ledger/:bucket/:key/:value", func(c *fiber.Ctx) error {
bucket := c.Params("bucket")
key := c.Params("key")
value := c.Params("value")
ledger, err := e.Ledger()
if err != nil {
return err
}
ledger.Persist(context.Background(), DefaultInterval, Timeout, bucket, key, value)
return c.JSON(announcing)
})
// Delete data from ledger
app.Get("ledger/:bucket", func(c *fiber.Ctx) error {
bucket := c.Params("bucket")
ledger, err := e.Ledger()
if err != nil {
return err
}
ledger.AnnounceDeleteBucket(context.Background(), DefaultInterval, Timeout, bucket)
return c.JSON(announcing)
})
app.Get("ledger/:bucket/:key", func(c *fiber.Ctx) error {
bucket := c.Params("bucket")
key := c.Params("key")
ledger, err := e.Ledger()
if err != nil {
return err
}
ledger.AnnounceDeleteBucketKey(context.Background(), DefaultInterval, Timeout, bucket, key)
return c.JSON(announcing)
})
}

12
core/p2p/federated_server.go
View file

@ -10,12 +10,18 @@ import (
"io"
"net"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/edgevpn/pkg/node"
"github.com/rs/zerolog/log"
)
func (f *FederatedServer) Start(ctx context.Context) error {
n, err := NewNode(f.p2ptoken)
func (f *FederatedServer) Start(ctx context.Context, p2pCommonFlags cliP2P.P2PCommonFlags) error {
p2pCfg := NewP2PConfig(p2pCommonFlags)
p2pCfg.NetworkToken = f.p2ptoken
p2pCfg.PeerGuard.Autocleanup = true
p2pCfg.PeerGuard.PeerGate = true
n, err := NewNode(p2pCfg)
if err != nil {
return fmt.Errorf("creating a new node: %w", err)
}
@ -24,7 +30,7 @@ func (f *FederatedServer) Start(ctx context.Context) error {
return fmt.Errorf("creating a new node: %w", err)
}
if err := ServiceDiscoverer(ctx, n, f.p2ptoken, f.service, func(servicesID string, tunnel NodeData) {
if err := ServiceDiscoverer(ctx, n, f.service, func(servicesID string, tunnel NodeData) {
log.Debug().Msgf("Discovered node: %s", tunnel.ID)
}, false); err != nil {
return err

169
core/p2p/p2p.go
View file

@ -5,32 +5,35 @@ package p2p
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"os"
"strings"
"sync"
"time"
"github.com/ipfs/go-log"
"github.com/libp2p/go-libp2p/core/crypto"
"github.com/libp2p/go-libp2p/core/peer"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
"github.com/mudler/LocalAI/pkg/utils"
"github.com/mudler/edgevpn/pkg/config"
p2pConfig "github.com/mudler/edgevpn/pkg/config"
"github.com/mudler/edgevpn/pkg/node"
"github.com/mudler/edgevpn/pkg/protocol"
"github.com/mudler/edgevpn/pkg/services"
"github.com/mudler/edgevpn/pkg/types"
eutils "github.com/mudler/edgevpn/pkg/utils"
"github.com/multiformats/go-multiaddr"
"github.com/phayes/freeport"
zlog "github.com/rs/zerolog/log"
"github.com/mudler/edgevpn/pkg/logger"
)
func generateNewConnectionData(DHTInterval, OTPInterval int) *node.YAMLConnectionConfig {
const DefaultInterval = 10 * time.Second
func GenerateNewConnectionData(DHTInterval, OTPInterval int, privkey string, peerguardMode bool) (*node.YAMLConnectionConfig, error) {
maxMessSize := 20 << 20 // 20MB
keyLength := 43
if DHTInterval == 0 {
@ -40,7 +43,7 @@ func generateNewConnectionData(DHTInterval, OTPInterval int) *node.YAMLConnectio
OTPInterval = 9000
}
return &node.YAMLConnectionConfig{
connectionConfig := node.YAMLConnectionConfig{
MaxMessageSize: maxMessSize,
RoomName: eutils.RandStringRunes(keyLength),
Rendezvous: eutils.RandStringRunes(keyLength),
@ -58,11 +61,27 @@ func generateNewConnectionData(DHTInterval, OTPInterval int) *node.YAMLConnectio
},
},
}
}
func GenerateToken(DHTInterval, OTPInterval int) string {
// Generates a new config and exit
return generateNewConnectionData(DHTInterval, OTPInterval).Base64()
if peerguardMode {
key, err := crypto.UnmarshalPrivateKey([]byte(privkey))
if err != nil {
return &connectionConfig, err
}
pid, err := peer.IDFromPublicKey(key.GetPublic())
if err != nil {
return &connectionConfig, err
}
connectionConfig.TrustedPeerIDS = []string{
pid.String(),
}
connectionConfig.ProtectedStoreKeys = []string{
"trustzone",
"trustzoneAuth",
}
}
return &connectionConfig, nil
}
func IsP2PEnabled() bool {
@ -176,11 +195,11 @@ func allocateLocalService(ctx context.Context, node *node.Node, listenAddr, serv
// This is the main of the server (which keeps the env variable updated)
// This starts a goroutine that keeps LLAMACPP_GRPC_SERVERS updated with the discovered services
func ServiceDiscoverer(ctx context.Context, n *node.Node, token, servicesID string, discoveryFunc func(serviceID string, node NodeData), allocate bool) error {
func ServiceDiscoverer(ctx context.Context, n *node.Node, servicesID string, discoveryFunc func(serviceID string, node NodeData), allocate bool) error {
if servicesID == "" {
servicesID = defaultServicesID
}
tunnels, err := discoveryTunnels(ctx, n, token, servicesID, allocate)
tunnels, err := discoveryTunnels(ctx, n, servicesID, allocate)
if err != nil {
return err
}
@ -207,7 +226,7 @@ func ServiceDiscoverer(ctx context.Context, n *node.Node, token, servicesID stri
return nil
}
func discoveryTunnels(ctx context.Context, n *node.Node, token, servicesID string, allocate bool) (chan NodeData, error) {
func discoveryTunnels(ctx context.Context, n *node.Node, servicesID string, allocate bool) (chan NodeData, error) {
tunnels := make(chan NodeData)
ledger, err := n.Ledger()
@ -316,16 +335,16 @@ func ensureService(ctx context.Context, n *node.Node, nd *NodeData, sserv string
}
// This is the P2P worker main
func ExposeService(ctx context.Context, host, port, token, servicesID string) (*node.Node, error) {
if servicesID == "" {
servicesID = defaultServicesID
}
func ExposeService(ctx context.Context, p2pCfg p2pConfig.Config, host, port, servicesID string) (*node.Node, error) {
llger := logger.New(log.LevelFatal)
nodeOpts, err := newNodeOpts(token)
nodeOpts, _, err := p2pCfg.ToOpts(llger)
if err != nil {
return nil, err
return nil, fmt.Errorf("parsing config for new node: %w", err)
}
nodeOpts = append(nodeOpts, node.FromBase64(true, p2pCfg.Discovery.DHT, p2pCfg.NetworkToken, nil, nil))
nodeOpts = append(nodeOpts, services.Alive(30*time.Second, 900*time.Second, 15*time.Minute)...)
// generate a random string for the name
name := utils.RandString(10)
@ -347,6 +366,9 @@ func ExposeService(ctx context.Context, host, port, token, servicesID string) (*
return n, fmt.Errorf("creating a new node: %w", err)
}
if servicesID == "" {
servicesID = defaultServicesID
}
ledger.Announce(
ctx,
20*time.Second,
@ -364,11 +386,15 @@ func ExposeService(ctx context.Context, host, port, token, servicesID string) (*
return n, err
}
func NewNode(token string) (*node.Node, error) {
nodeOpts, err := newNodeOpts(token)
func NewNode(p2pCfg p2pConfig.Config) (*node.Node, error) {
llger := logger.New(log.LevelFatal)
nodeOpts, _, err := p2pCfg.ToOpts(llger)
if err != nil {
return nil, err
return nil, fmt.Errorf("parsing config for new node: %w", err)
}
nodeOpts = append(nodeOpts, node.FromBase64(true, p2pCfg.Discovery.DHT, p2pCfg.NetworkToken, nil, nil))
nodeOpts = append(nodeOpts, services.Alive(30*time.Second, 900*time.Second, 15*time.Minute)...)
n, err := node.New(nodeOpts...)
if err != nil {
@ -378,89 +404,64 @@ func NewNode(token string) (*node.Node, error) {
return n, nil
}
func newNodeOpts(token string) ([]node.Option, error) {
llger := logger.New(log.LevelFatal)
defaultInterval := 10 * time.Second
func NewP2PConfig(p2pCommonFlags cliP2P.P2PCommonFlags) p2pConfig.Config {
pa := p2pCommonFlags.Peer2PeerAuthProvders
d := map[string]map[string]interface{}{}
json.Unmarshal([]byte(pa), &d)
// TODO: move this up, expose more config options when creating a node
noDHT := os.Getenv("LOCALAI_P2P_DISABLE_DHT") == "true"
noLimits := os.Getenv("LOCALAI_P2P_ENABLE_LIMITS") == "true"
var listenMaddrs []string
var bootstrapPeers []string
laddrs := os.Getenv("LOCALAI_P2P_LISTEN_MADDRS")
if laddrs != "" {
listenMaddrs = strings.Split(laddrs, ",")
}
bootmaddr := os.Getenv("LOCALAI_P2P_BOOTSTRAP_PEERS_MADDRS")
if bootmaddr != "" {
bootstrapPeers = strings.Split(bootmaddr, ",")
}
dhtAnnounceMaddrs := stringsToMultiAddr(strings.Split(os.Getenv("LOCALAI_P2P_DHT_ANNOUNCE_MADDRS"), ","))
libp2ploglevel := os.Getenv("LOCALAI_P2P_LIB_LOGLEVEL")
if libp2ploglevel == "" {
libp2ploglevel = "fatal"
}
c := config.Config{
ListenMaddrs: listenMaddrs,
DHTAnnounceMaddrs: dhtAnnounceMaddrs,
Limit: config.ResourceLimit{
Enable: noLimits,
c := p2pConfig.Config{
ListenMaddrs: p2pCommonFlags.Peer2PeerListenAddrs,
DHTAnnounceMaddrs: utils.StringsToMultiAddr(p2pCommonFlags.Peer2PeerDHTAnnounceAddrs),
Limit: p2pConfig.ResourceLimit{
Enable: p2pCommonFlags.Peer2PeerLimit,
MaxConns: 100,
},
NetworkToken: token,
LowProfile: false,
LogLevel: logLevel,
Libp2pLogLevel: libp2ploglevel,
Ledger: config.Ledger{
SyncInterval: defaultInterval,
AnnounceInterval: defaultInterval,
LowProfile: false,
LogLevel: logLevel,
Ledger: p2pConfig.Ledger{
SyncInterval: DefaultInterval,
AnnounceInterval: DefaultInterval,
},
NAT: config.NAT{
NAT: p2pConfig.NAT{
Service: true,
Map: true,
RateLimit: true,
RateLimitGlobal: 100,
RateLimitPeer: 100,
RateLimitInterval: defaultInterval,
RateLimitInterval: DefaultInterval,
},
Discovery: config.Discovery{
DHT: !noDHT,
Discovery: p2pConfig.Discovery{
DHT: !p2pCommonFlags.Peer2PeerNoDHT,
MDNS: true,
Interval: 10 * time.Second,
BootstrapPeers: bootstrapPeers,
Interval: DefaultInterval,
BootstrapPeers: p2pCommonFlags.Peer2PeerBootAddrs,
},
Connection: config.Connection{
Connection: p2pConfig.Connection{
HolePunch: true,
AutoRelay: true,
MaxConnections: 1000,
},
PeerGuard: p2pConfig.PeerGuard{
Enable: p2pCommonFlags.Peer2PeerUsePeerguard,
// Default from edgevpn
SyncInterval: 120 * time.Second,
AuthProviders: d,
},
}
nodeOpts, _, err := c.ToOpts(llger)
if err != nil {
return nil, fmt.Errorf("parsing options: %w", err)
privkey := p2pCommonFlags.Peer2PeerPrivkey
if privkey != "" {
c.Privkey = []byte(privkey)
}
nodeOpts = append(nodeOpts, services.Alive(30*time.Second, 900*time.Second, 15*time.Minute)...)
return nodeOpts, nil
}
func stringsToMultiAddr(peers []string) []multiaddr.Multiaddr {
res := []multiaddr.Multiaddr{}
for _, p := range peers {
addr, err := multiaddr.NewMultiaddr(p)
if err != nil {
continue
}
res = append(res, addr)
libp2ploglevel := p2pCommonFlags.Peer2PeerLibLoglevel
if libp2ploglevel == "" {
libp2ploglevel = "fatal"
}
return res
c.Libp2pLogLevel = libp2ploglevel
return c
}
func copyStream(closer chan struct{}, dst io.Writer, src io.Reader) {

18
core/p2p/p2p_disabled.go
View file

@ -7,22 +7,24 @@ import (
"context"
"fmt"
cliP2P "github.com/mudler/LocalAI/core/cli/p2p"
p2pConfig "github.com/mudler/edgevpn/pkg/config"
"github.com/mudler/edgevpn/pkg/node"
)
func GenerateToken(DHTInterval, OTPInterval int) string {
return "not implemented"
func GenerateNewConnectionData(DHTInterval, OTPInterval int, privkey string, peerguardMode bool) (*node.YAMLConnectionConfig, error) {
return nil, fmt.Errorf("not implemented")
}
func (f *FederatedServer) Start(ctx context.Context) error {
func (f *FederatedServer) Start(ctx context.Context, p2pCommonFlags cliP2P.P2PCommonFlags) error {
return fmt.Errorf("not implemented")
}
func ServiceDiscoverer(ctx context.Context, node *node.Node, token, servicesID string, fn func(string, NodeData), allocate bool) error {
func ServiceDiscoverer(ctx context.Context, n *node.Node, servicesID string, discoveryFunc func(serviceID string, node NodeData), allocate bool) error {
return fmt.Errorf("not implemented")
}
func ExposeService(ctx context.Context, host, port, token, servicesID string) (*node.Node, error) {
func ExposeService(ctx context.Context, p2pCfg p2pConfig.Config, host, port, servicesID string) (*node.Node, error) {
return nil, fmt.Errorf("not implemented")
}
@ -30,6 +32,10 @@ func IsP2PEnabled() bool {
return false
}
func NewNode(token string) (*node.Node, error) {
func NewNode(p2pCfg p2pConfig.Config) (*node.Node, error) {
return nil, fmt.Errorf("not implemented")
}
func NewP2PConfig(p2pCommonFlags cliP2P.P2PCommonFlags) p2pConfig.Config {
return p2pConfig.Config{}
}