migrated more cluster

cluster/cluster.go

@@ -21,7 +21,9 @@ package cluster
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"regexp"
 	"runtime"
 	"sync"
@@ -66,6 +68,10 @@ type Cluster struct {
 	mux    sync.RWMutex
 	status atomic.Int32
 	socket *socket.Socket
+	client *http.Client
+
+	authTokenMux sync.RWMutex
+	authToken    *ClusterToken
 }
 
 func NewCluster(
@@ -78,6 +84,8 @@ func NewCluster(
 
 		storageManager: storageManager,
 		storages:       storages,
+
+		client: &http.Client{},
 	}
 	return
 }
@@ -87,6 +95,11 @@ func (cr *Cluster) ID() string {
 	return cr.opts.Id
 }
 
+// Secret returns the cluster secret
+func (cr *Cluster) Secret() string {
+	return cr.opts.Secret
+}
+
 // Host returns the cluster public host
 func (cr *Cluster) Host() string {
 	return cr.gcfg.Host
@@ -181,7 +194,7 @@ func (cr *Cluster) enable(ctx context.Context) error {
 			if msg, ok := ero["message"].(string); ok {
 				if hashMismatch := reFileHashMismatchError.FindStringSubmatch(msg); hashMismatch != nil {
 					hash := hashMismatch[1]
-					log.Warnf(Tr("warn.cluster.detected.hash.mismatch"), hash)
+					log.TrWarnf("warn.cluster.detected.hash.mismatch", hash)
 					cr.storageManager.RemoveForAll(hash)
 				}
 				return fmt.Errorf("Enable failed: %v", msg)
@@ -239,6 +252,7 @@ func (cr *Cluster) reEnable(disableSignal <-chan struct{}) {
 				select {
 				case <-ctx.Done():
 				case <-disableSignal:
+					timer.Stop()
 					cancel()
 				}
 			}()

cluster/config.go

@@ -86,7 +86,7 @@ func (cr *Cluster) fetchToken(ctx context.Context) (token *ClusterToken, err err
 		}
 	}()
 	req, err := cr.makeReq(ctx, http.MethodGet, "/openbmclapi-agent/challenge", url.Values{
-		"clusterId": {cr.clusterId},
+		"clusterId": {cr.ID()},
 	})
 	if err != nil {
 		return
@@ -110,7 +110,7 @@ func (cr *Cluster) fetchToken(ctx context.Context) (token *ClusterToken, err err
 	}
 
 	var buf [32]byte
-	hs := hmac.New(crypto.SHA256.New, ([]byte)(cr.clusterSecret))
+	hs := hmac.New(crypto.SHA256.New, ([]byte)(cr.Secret()))
 	hs.Write(([]byte)(res1.Challenge))
 	signature := hex.EncodeToString(hs.Sum(buf[:0]))
 
@@ -119,7 +119,7 @@ func (cr *Cluster) fetchToken(ctx context.Context) (token *ClusterToken, err err
 		Challenge string `json:"challenge"`
 		Signature string `json:"signature"`
 	}{
-		ClusterId: cr.clusterId,
+		ClusterId: cr.ID(),
 		Challenge: res1.Challenge,
 		Signature: signature,
 	})
@@ -159,7 +159,7 @@ func (cr *Cluster) refreshToken(ctx context.Context, oldToken string) (token *Cl
 		ClusterId string `json:"clusterId"`
 		Token     string `json:"token"`
 	}{
-		ClusterId: cr.clusterId,
+		ClusterId: cr.ID(),
 		Token:     oldToken,
 	})
 	if err != nil {

cluster/handler.go

@@ -21,29 +21,30 @@ package cluster
 
 import (
 	"net/http"
+
+	"github.com/LiterMC/go-openbmclapi/log"
+	"github.com/LiterMC/go-openbmclapi/storage"
 )
 
-func (cr *Cluster) HandleFile(req *http.Request, rw http.ResponseWriter, hash string) {
+func (cr *Cluster) HandleFile(req *http.Request, rw http.ResponseWriter, hash string, size int64) {
+	defer log.RecoverPanic(nil)
+	var err error
 	if cr.storageManager.ForEachFromRandom(cr.storages, func(s storage.Storage) bool {
-		log.Debugf("[handler]: Checking %s on storage [%d] %s ...", hash, i, sto.String())
+		opts := s.Options()
+		log.Debugf("[handler]: Checking %s on storage %s ...", hash, opts.Id)
 
-		sz, er := sto.ServeDownload(rw, req, hash, size)
+		sz, er := s.ServeDownload(rw, req, hash, size)
 		if er != nil {
-			log.Debugf("[handler]: File %s failed on storage [%d] %s: %v", hash, i, sto.String(), er)
+			log.Debugf("[handler]: File %s failed on storage %s: %v", hash, opts.Id, er)
 			err = er
 			return false
 		}
 		if sz >= 0 {
-			opts := cr.storageOpts[i]
-			cr.AddHits(1, sz, s.Options().Id)
-			if !keepaliveRec {
-				cr.statOnlyHits.Add(1)
-				cr.statOnlyHbts.Add(sz)
-			}
+			cr.AddHits(1, sz, opts.Id)
 		}
 		return true
 	}) {
 		return
 	}
-	http.Error(http.StatusInternation)
+	http.Error(rw, err.Error(), http.StatusInternalServerError)
 }

cluster/http.go

@@ -0,0 +1,25 @@
+/**
+ * OpenBmclAPI (Golang Edition)
+ * Copyright (C) 2024 Kevin Z <zyxkad@gmail.com>
+ * All rights reserved
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU Affero General Public License as published
+ *  by the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU Affero General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Affero General Public License
+ *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package cluster
+
+import (
+	"net/http"
+	"net/url"
+)

go.mod

@@ -1,6 +1,6 @@
 module github.com/LiterMC/go-openbmclapi
 
-go 1.21.6
+go 1.22.0
 
 require (
 	github.com/LiterMC/socket.io v0.2.4

log/tr.go

@@ -38,7 +38,3 @@ func TrWarnf(key string, vals ...any) {
 func TrErrorf(key string, vals ...any) {
 	Errorf(lang.Tr(key), vals...)
 }
-
-func TrPanicf(key string, vals ...any) {
-	Panicf(lang.Tr(key), vals...)
-}

storage/manager.go

@@ -20,6 +20,8 @@
 package storage
 
 import (
+	"errors"
+
 	"github.com/LiterMC/go-openbmclapi/log"
 	"github.com/LiterMC/go-openbmclapi/utils"
 )
@@ -146,3 +148,17 @@ func forEachFromRandomIndexWithPossibility(poss []uint, total uint, cb func(i in
 	}
 	return false
 }
+
+func (m *Manager) RemoveForAll(hash string) error {
+	errCh := make(chan error, 0)
+	for _, s := range m.Storages {
+		go func(s Storage) {
+			errCh <- s.Remove(hash)
+		}(s)
+	}
+	errs := make([]error, len(m.Storages))
+	for i := range len(m.Storages) {
+		errs[i] = <-errCh
+	}
+	return errors.Join(errs...)
+}