mirror of
https://github.com/photoprism/photoprism.git
synced 2025-12-12 08:44:04 +01:00
Config: Add cluster instance bootstrap and registration hook #98
Signed-off-by: Michael Mayer <michael@photoprism.app>
This commit is contained in:
64
internal/commands/cluster_exit_codes_test.go
Normal file
64
internal/commands/cluster_exit_codes_test.go
Normal file
@@ -0,0 +1,64 @@
|
||||
package commands
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/urfave/cli/v2"
|
||||
)
|
||||
|
||||
func TestExitCodes_Register_ValidationAndUnauthorized(t *testing.T) {
|
||||
t.Run("MissingURL", func(t *testing.T) {
|
||||
ctx := NewTestContext([]string{"register", "--name", "pp-node-01", "--type", "instance", "--portal-token", "token"})
|
||||
err := ClusterRegisterCommand.Action(ctx)
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestExitCodes_Nodes_PortalOnlyMisuse(t *testing.T) {
|
||||
t.Run("ListNotPortal", func(t *testing.T) {
|
||||
ctx := NewTestContext([]string{"ls"})
|
||||
err := ClusterNodesListCommand.Action(ctx)
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
})
|
||||
t.Run("ShowNotPortal", func(t *testing.T) {
|
||||
ctx := NewTestContext([]string{"show", "any"})
|
||||
err := ClusterNodesShowCommand.Action(ctx)
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
})
|
||||
t.Run("RemoveNotPortal", func(t *testing.T) {
|
||||
ctx := NewTestContext([]string{"rm", "any"})
|
||||
err := ClusterNodesRemoveCommand.Action(ctx)
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
})
|
||||
t.Run("ModNotPortal", func(t *testing.T) {
|
||||
ctx := NewTestContext([]string{"mod", "any", "--type", "instance", "-y"})
|
||||
err := ClusterNodesModCommand.Action(ctx)
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -37,17 +37,17 @@ var ClusterNodesListCommand = &cli.Command{
|
||||
func clusterNodesListAction(ctx *cli.Context) error {
|
||||
return CallWithDependencies(ctx, func(conf *config.Config) error {
|
||||
if !conf.IsPortal() {
|
||||
return fmt.Errorf("node listing is only available on a Portal node")
|
||||
return cli.Exit(fmt.Errorf("node listing is only available on a Portal node"), 2)
|
||||
}
|
||||
|
||||
r, err := reg.NewFileRegistry(conf)
|
||||
if err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
items, err := r.List()
|
||||
if err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
// Pagination identical to API defaults.
|
||||
@@ -97,7 +97,10 @@ func clusterNodesListAction(ctx *cli.Context) error {
|
||||
|
||||
result, err := report.RenderFormat(rows, cols, report.CliFormat(ctx))
|
||||
fmt.Printf("\n%s\n", result)
|
||||
return err
|
||||
if err != nil {
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -31,29 +31,29 @@ var ClusterNodesModCommand = &cli.Command{
|
||||
func clusterNodesModAction(ctx *cli.Context) error {
|
||||
return CallWithDependencies(ctx, func(conf *config.Config) error {
|
||||
if !conf.IsPortal() {
|
||||
return fmt.Errorf("node update is only available on a Portal node")
|
||||
return cli.Exit(fmt.Errorf("node update is only available on a Portal node"), 2)
|
||||
}
|
||||
|
||||
key := ctx.Args().First()
|
||||
if key == "" {
|
||||
return cli.ShowSubcommandHelp(ctx)
|
||||
return cli.Exit(fmt.Errorf("node id or name is required"), 2)
|
||||
}
|
||||
|
||||
r, err := reg.NewFileRegistry(conf)
|
||||
if err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
n, getErr := r.Get(key)
|
||||
if getErr != nil {
|
||||
name := clean.TypeLowerDash(key)
|
||||
if name == "" {
|
||||
return fmt.Errorf("invalid node identifier")
|
||||
return cli.Exit(fmt.Errorf("invalid node identifier"), 2)
|
||||
}
|
||||
n, getErr = r.FindByName(name)
|
||||
}
|
||||
if getErr != nil || n == nil {
|
||||
return fmt.Errorf("node not found")
|
||||
return cli.Exit(fmt.Errorf("node not found"), 3)
|
||||
}
|
||||
|
||||
if v := ctx.String("type"); v != "" {
|
||||
@@ -83,7 +83,7 @@ func clusterNodesModAction(ctx *cli.Context) error {
|
||||
}
|
||||
|
||||
if err := r.Put(n); err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
log.Infof("node %s has been updated", clean.LogQuote(n.Name))
|
||||
|
||||
@@ -25,17 +25,17 @@ var ClusterNodesRemoveCommand = &cli.Command{
|
||||
func clusterNodesRemoveAction(ctx *cli.Context) error {
|
||||
return CallWithDependencies(ctx, func(conf *config.Config) error {
|
||||
if !conf.IsPortal() {
|
||||
return fmt.Errorf("node delete is only available on a Portal node")
|
||||
return cli.Exit(fmt.Errorf("node delete is only available on a Portal node"), 2)
|
||||
}
|
||||
|
||||
key := ctx.Args().First()
|
||||
if key == "" {
|
||||
return cli.ShowSubcommandHelp(ctx)
|
||||
return cli.Exit(fmt.Errorf("node id or name is required"), 2)
|
||||
}
|
||||
|
||||
r, err := reg.NewFileRegistry(conf)
|
||||
if err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
// Resolve to id for deletion, but also support name.
|
||||
@@ -44,7 +44,7 @@ func clusterNodesRemoveAction(ctx *cli.Context) error {
|
||||
if n, err2 := r.FindByName(clean.TypeLowerDash(key)); err2 == nil && n != nil {
|
||||
id = n.ID
|
||||
} else {
|
||||
return fmt.Errorf("node not found")
|
||||
return cli.Exit(fmt.Errorf("node not found"), 3)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -58,7 +58,7 @@ func clusterNodesRemoveAction(ctx *cli.Context) error {
|
||||
}
|
||||
|
||||
if err := r.Delete(id); err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
log.Infof("node %s has been deleted", clean.Log(id))
|
||||
|
||||
@@ -35,7 +35,7 @@ func clusterNodesRotateAction(ctx *cli.Context) error {
|
||||
return CallWithDependencies(ctx, func(conf *config.Config) error {
|
||||
key := ctx.Args().First()
|
||||
if key == "" {
|
||||
return cli.ShowSubcommandHelp(ctx)
|
||||
return cli.Exit(fmt.Errorf("node id or name is required"), 2)
|
||||
}
|
||||
|
||||
// Determine node name. On portal, resolve id->name via registry; otherwise treat key as name.
|
||||
@@ -50,7 +50,7 @@ func clusterNodesRotateAction(ctx *cli.Context) error {
|
||||
}
|
||||
}
|
||||
if name == "" {
|
||||
return fmt.Errorf("invalid node identifier")
|
||||
return cli.Exit(fmt.Errorf("invalid node identifier"), 2)
|
||||
}
|
||||
|
||||
// Portal URL and token
|
||||
@@ -62,7 +62,7 @@ func clusterNodesRotateAction(ctx *cli.Context) error {
|
||||
portalURL = os.Getenv(config.EnvVar("portal-url"))
|
||||
}
|
||||
if portalURL == "" {
|
||||
return fmt.Errorf("portal URL is required (use --portal-url or set portal-url)")
|
||||
return cli.Exit(fmt.Errorf("portal URL is required (use --portal-url or set portal-url)"), 2)
|
||||
}
|
||||
token := ctx.String("portal-token")
|
||||
if token == "" {
|
||||
@@ -72,7 +72,7 @@ func clusterNodesRotateAction(ctx *cli.Context) error {
|
||||
token = os.Getenv(config.EnvVar("portal-token"))
|
||||
}
|
||||
if token == "" {
|
||||
return fmt.Errorf("portal token is required (use --portal-token or set portal-token)")
|
||||
return cli.Exit(fmt.Errorf("portal token is required (use --portal-token or set portal-token)"), 2)
|
||||
}
|
||||
|
||||
// Default: rotate DB only if no flag given (safer default)
|
||||
@@ -107,7 +107,22 @@ func clusterNodesRotateAction(ctx *cli.Context) error {
|
||||
url := stringsTrimRightSlash(portalURL) + "/api/v1/cluster/nodes/register"
|
||||
var resp cluster.RegisterResponse
|
||||
if err := postWithBackoff(url, token, b, &resp); err != nil {
|
||||
return err
|
||||
// Map common HTTP errors similarly to register command
|
||||
if he, ok := err.(*httpError); ok {
|
||||
switch he.Status {
|
||||
case 401, 403:
|
||||
return cli.Exit(fmt.Errorf("%s", he.Error()), 4)
|
||||
case 409:
|
||||
return cli.Exit(fmt.Errorf("%s", he.Error()), 5)
|
||||
case 400:
|
||||
return cli.Exit(fmt.Errorf("%s", he.Error()), 2)
|
||||
case 404:
|
||||
return cli.Exit(fmt.Errorf("%s", he.Error()), 3)
|
||||
case 429:
|
||||
return cli.Exit(fmt.Errorf("%s", he.Error()), 6)
|
||||
}
|
||||
}
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
if ctx.Bool("json") {
|
||||
|
||||
@@ -24,17 +24,17 @@ var ClusterNodesShowCommand = &cli.Command{
|
||||
func clusterNodesShowAction(ctx *cli.Context) error {
|
||||
return CallWithDependencies(ctx, func(conf *config.Config) error {
|
||||
if !conf.IsPortal() {
|
||||
return fmt.Errorf("node show is only available on a Portal node")
|
||||
return cli.Exit(fmt.Errorf("node show is only available on a Portal node"), 2)
|
||||
}
|
||||
|
||||
key := ctx.Args().First()
|
||||
if key == "" {
|
||||
return cli.ShowSubcommandHelp(ctx)
|
||||
return cli.Exit(fmt.Errorf("node id or name is required"), 2)
|
||||
}
|
||||
|
||||
r, err := reg.NewFileRegistry(conf)
|
||||
if err != nil {
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
// Resolve by id first, then by normalized name.
|
||||
@@ -42,12 +42,12 @@ func clusterNodesShowAction(ctx *cli.Context) error {
|
||||
if getErr != nil {
|
||||
name := clean.TypeLowerDash(key)
|
||||
if name == "" {
|
||||
return fmt.Errorf("invalid node identifier")
|
||||
return cli.Exit(fmt.Errorf("invalid node identifier"), 2)
|
||||
}
|
||||
n, getErr = r.FindByName(name)
|
||||
}
|
||||
if getErr != nil || n == nil {
|
||||
return fmt.Errorf("node not found")
|
||||
return cli.Exit(fmt.Errorf("node not found"), 3)
|
||||
}
|
||||
|
||||
opts := reg.NodeOpts{IncludeInternalURL: true, IncludeDBMeta: true}
|
||||
@@ -67,6 +67,9 @@ func clusterNodesShowAction(ctx *cli.Context) error {
|
||||
rows := [][]string{{dto.ID, dto.Name, dto.Type, dto.InternalURL, dbName, dbUser, dbRot, dto.CreatedAt, dto.UpdatedAt}}
|
||||
out, err := report.RenderFormat(rows, cols, report.CliFormat(ctx))
|
||||
fmt.Printf("\n%s\n", out)
|
||||
return err
|
||||
if err != nil {
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/photoprism/photoprism/internal/service/cluster"
|
||||
"github.com/photoprism/photoprism/pkg/clean"
|
||||
"github.com/photoprism/photoprism/pkg/fs"
|
||||
"github.com/photoprism/photoprism/pkg/service/http/header"
|
||||
"github.com/photoprism/photoprism/pkg/txt/report"
|
||||
)
|
||||
|
||||
@@ -51,13 +52,13 @@ func clusterRegisterAction(ctx *cli.Context) error {
|
||||
name = clean.TypeLowerDash(conf.NodeName())
|
||||
}
|
||||
if name == "" {
|
||||
return fmt.Errorf("node name is required (use --name or set node-name)")
|
||||
return cli.Exit(fmt.Errorf("node name is required (use --name or set node-name)"), 2)
|
||||
}
|
||||
nodeType := clean.TypeLowerDash(ctx.String("type"))
|
||||
switch nodeType {
|
||||
case "instance", "service":
|
||||
default:
|
||||
return fmt.Errorf("invalid --type (must be instance or service)")
|
||||
return cli.Exit(fmt.Errorf("invalid --type (must be instance or service)"), 2)
|
||||
}
|
||||
|
||||
portalURL := ctx.String("portal-url")
|
||||
@@ -65,14 +66,14 @@ func clusterRegisterAction(ctx *cli.Context) error {
|
||||
portalURL = conf.PortalUrl()
|
||||
}
|
||||
if portalURL == "" {
|
||||
return fmt.Errorf("portal URL is required (use --portal-url or set portal-url)")
|
||||
return cli.Exit(fmt.Errorf("portal URL is required (use --portal-url or set portal-url)"), 2)
|
||||
}
|
||||
token := ctx.String("portal-token")
|
||||
if token == "" {
|
||||
token = conf.PortalToken()
|
||||
}
|
||||
if token == "" {
|
||||
return fmt.Errorf("portal token is required (use --portal-token or set portal-token)")
|
||||
return cli.Exit(fmt.Errorf("portal token is required (use --portal-token or set portal-token)"), 2)
|
||||
}
|
||||
|
||||
body := map[string]interface{}{
|
||||
@@ -91,22 +92,22 @@ func clusterRegisterAction(ctx *cli.Context) error {
|
||||
if err := postWithBackoff(url, token, b, &resp); err != nil {
|
||||
var httpErr *httpError
|
||||
if errors.As(err, &httpErr) && httpErr.Status == http.StatusTooManyRequests {
|
||||
return fmt.Errorf("portal rate-limited registration attempts")
|
||||
return cli.Exit(fmt.Errorf("portal rate-limited registration attempts"), 6)
|
||||
}
|
||||
// Map common errors
|
||||
if errors.As(err, &httpErr) {
|
||||
switch httpErr.Status {
|
||||
case http.StatusUnauthorized, http.StatusForbidden:
|
||||
return fmt.Errorf("%s", httpErr.Error())
|
||||
return cli.Exit(fmt.Errorf("%s", httpErr.Error()), 4)
|
||||
case http.StatusConflict:
|
||||
return fmt.Errorf("%s", httpErr.Error())
|
||||
return cli.Exit(fmt.Errorf("%s", httpErr.Error()), 5)
|
||||
case http.StatusBadRequest:
|
||||
return fmt.Errorf("%s", httpErr.Error())
|
||||
return cli.Exit(fmt.Errorf("%s", httpErr.Error()), 2)
|
||||
case http.StatusNotFound:
|
||||
return fmt.Errorf("%s", httpErr.Error())
|
||||
return cli.Exit(fmt.Errorf("%s", httpErr.Error()), 3)
|
||||
}
|
||||
}
|
||||
return err
|
||||
return cli.Exit(err, 1)
|
||||
}
|
||||
|
||||
// Output
|
||||
@@ -169,10 +170,11 @@ func postWithBackoff(url, token string, payload []byte, out any) error {
|
||||
delay := 500 * time.Millisecond
|
||||
for attempt := 0; attempt < 6; attempt++ {
|
||||
req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(payload))
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
header.SetAuthorization(req, token)
|
||||
req.Header.Set(header.ContentType, "application/json")
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
client := &http.Client{Timeout: cluster.BootstrapRegisterTimeout}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/tidwall/gjson"
|
||||
"github.com/urfave/cli/v2"
|
||||
|
||||
cfg "github.com/photoprism/photoprism/internal/config"
|
||||
)
|
||||
@@ -242,7 +243,11 @@ func TestClusterRegister_HTTPUnauthorized(t *testing.T) {
|
||||
_, err := RunWithTestContext(ClusterRegisterCommand, []string{
|
||||
"register", "--name", "pp-node-unauth", "--type", "instance", "--portal-url", ts.URL, "--portal-token", "wrong", "--json",
|
||||
})
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 4, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterRegister_HTTPConflict(t *testing.T) {
|
||||
@@ -254,7 +259,11 @@ func TestClusterRegister_HTTPConflict(t *testing.T) {
|
||||
_, err := RunWithTestContext(ClusterRegisterCommand, []string{
|
||||
"register", "--name", "pp-node-conflict", "--type", "instance", "--portal-url", ts.URL, "--portal-token", "test-token", "--json",
|
||||
})
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 5, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterRegister_HTTPBadRequest(t *testing.T) {
|
||||
@@ -266,7 +275,11 @@ func TestClusterRegister_HTTPBadRequest(t *testing.T) {
|
||||
_, err := RunWithTestContext(ClusterRegisterCommand, []string{
|
||||
"register", "--name", "pp node invalid", "--type", "instance", "--portal-url", ts.URL, "--portal-token", "test-token", "--json",
|
||||
})
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterRegister_HTTPRateLimitOnceThenOK(t *testing.T) {
|
||||
@@ -304,7 +317,11 @@ func TestClusterNodesRotate_HTTPUnauthorized_JSON(t *testing.T) {
|
||||
_, err := RunWithTestContext(ClusterNodesRotateCommand, []string{
|
||||
"rotate", "--json", "--portal-url=" + ts.URL, "--portal-token=wrong", "--db", "--yes", "pp-node-x",
|
||||
})
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 4, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterNodesRotate_HTTPConflict_JSON(t *testing.T) {
|
||||
@@ -316,7 +333,11 @@ func TestClusterNodesRotate_HTTPConflict_JSON(t *testing.T) {
|
||||
_, err := RunWithTestContext(ClusterNodesRotateCommand, []string{
|
||||
"rotate", "--json", "--portal-url=" + ts.URL, "--portal-token=test-token", "--db", "--yes", "pp-node-x",
|
||||
})
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 5, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterNodesRotate_HTTPBadRequest_JSON(t *testing.T) {
|
||||
@@ -328,7 +349,11 @@ func TestClusterNodesRotate_HTTPBadRequest_JSON(t *testing.T) {
|
||||
_, err := RunWithTestContext(ClusterNodesRotateCommand, []string{
|
||||
"rotate", "--json", "--portal-url=" + ts.URL, "--portal-token=test-token", "--db", "--yes", "pp node invalid",
|
||||
})
|
||||
assert.Error(t, err)
|
||||
if ec, ok := err.(cli.ExitCoder); ok {
|
||||
assert.Equal(t, 2, ec.ExitCode())
|
||||
} else {
|
||||
t.Fatalf("expected ExitCoder, got %T", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterNodesRotate_HTTPRateLimitOnceThenOK_JSON(t *testing.T) {
|
||||
|
||||
@@ -2,7 +2,6 @@ package commands
|
||||
|
||||
import (
|
||||
"archive/zip"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -13,8 +12,10 @@ import (
|
||||
"github.com/urfave/cli/v2"
|
||||
|
||||
"github.com/photoprism/photoprism/internal/config"
|
||||
"github.com/photoprism/photoprism/internal/service/cluster"
|
||||
"github.com/photoprism/photoprism/pkg/clean"
|
||||
"github.com/photoprism/photoprism/pkg/fs"
|
||||
"github.com/photoprism/photoprism/pkg/service/http/header"
|
||||
)
|
||||
|
||||
// ClusterThemePullCommand downloads the Portal theme and installs it.
|
||||
@@ -89,6 +90,8 @@ func clusterThemePullAction(ctx *cli.Context) error {
|
||||
|
||||
// Download zip to a temp file.
|
||||
zipURL := portalURL + "/api/v1/cluster/theme"
|
||||
// TODO: Enforce TLS for non-local Portal URLs (similar to bootstrap) unless an explicit
|
||||
// insecure override is provided. Consider adding a --tls-only / --insecure flag.
|
||||
tmpFile, err := os.CreateTemp("", "photoprism-theme-*.zip")
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -101,8 +104,12 @@ func clusterThemePullAction(ctx *cli.Context) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
header.SetAuthorization(req, token)
|
||||
req.Header.Set(header.Accept, header.ContentTypeZip)
|
||||
|
||||
// Use a short timeout for responsiveness; align with bootstrap defaults.
|
||||
client := &http.Client{Timeout: cluster.BootstrapRegisterTimeout}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -111,13 +118,15 @@ func clusterThemePullAction(ctx *cli.Context) error {
|
||||
// Map common codes to clearer messages
|
||||
switch resp.StatusCode {
|
||||
case http.StatusUnauthorized, http.StatusForbidden:
|
||||
return fmt.Errorf("unauthorized; check portal token and permissions (%s)", resp.Status)
|
||||
return cli.Exit(fmt.Errorf("unauthorized; check portal token and permissions (%s)", resp.Status), 4)
|
||||
case http.StatusTooManyRequests:
|
||||
return fmt.Errorf("rate limited by portal (%s)", resp.Status)
|
||||
return cli.Exit(fmt.Errorf("rate limited by portal (%s)", resp.Status), 6)
|
||||
case http.StatusNotFound:
|
||||
return fmt.Errorf("portal theme not found (%s)", resp.Status)
|
||||
return cli.Exit(fmt.Errorf("portal theme not found (%s)", resp.Status), 3)
|
||||
case http.StatusBadRequest:
|
||||
return cli.Exit(fmt.Errorf("bad request (%s)", resp.Status), 2)
|
||||
default:
|
||||
return fmt.Errorf("download failed: %s", resp.Status)
|
||||
return cli.Exit(fmt.Errorf("download failed: %s", resp.Status), 1)
|
||||
}
|
||||
}
|
||||
if _, err = io.Copy(tmpFile, resp.Body); err != nil {
|
||||
@@ -174,9 +183,7 @@ func unzipSafe(zipPath, dest string) error {
|
||||
return err
|
||||
}
|
||||
defer r.Close()
|
||||
if len(r.File) == 0 {
|
||||
return errors.New("theme archive is empty")
|
||||
}
|
||||
// Empty theme archives are valid; install succeeds without files.
|
||||
for _, f := range r.File {
|
||||
// Directories are indicated by trailing '/'; ensure canonical path
|
||||
name := filepath.Clean(f.Name)
|
||||
|
||||
@@ -79,8 +79,12 @@ func RunWithTestContext(cmd *cli.Command, args []string) (output string, err err
|
||||
// a nil pointer panic in the "github.com/urfave/cli/v2" package.
|
||||
cmd.HideHelp = true
|
||||
|
||||
// Run command with test context.
|
||||
// Run command via cli.Command.Run but neutralize os.Exit so ExitCoder
|
||||
// errors don't terminate the test binary.
|
||||
output = capture.Output(func() {
|
||||
origExiter := cli.OsExiter
|
||||
cli.OsExiter = func(int) {}
|
||||
defer func() { cli.OsExiter = origExiter }()
|
||||
err = cmd.Run(ctx, args...)
|
||||
})
|
||||
|
||||
|
||||
@@ -237,6 +237,11 @@ func (c *Config) Init() error {
|
||||
// Load settings from the "settings.yml" config file.
|
||||
c.initSettings()
|
||||
|
||||
// Initialize early extensions before connecting to the database so they can
|
||||
// influence DB settings (e.g., cluster bootstrap providing MariaDB creds).
|
||||
log.Debugf("config: initializing early extensions")
|
||||
EarlyExt().InitEarly(c)
|
||||
|
||||
// Connect to database.
|
||||
if err := c.connectDb(); err != nil {
|
||||
return err
|
||||
|
||||
@@ -56,6 +56,9 @@ func TestConfig_BackupDatabase(t *testing.T) {
|
||||
|
||||
func TestConfig_BackupDatabasePath(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
// Ensure DB defaults (SQLite) so path resolves to sqlite backup path
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Contains(t, c.BackupDatabasePath(), "/storage/testdata/backup/sqlite")
|
||||
}
|
||||
|
||||
|
||||
@@ -34,6 +34,21 @@ func TestConfig_Cluster(t *testing.T) {
|
||||
// Use an isolated config path so we don't affect repo storage fixtures.
|
||||
tempCfg := t.TempDir()
|
||||
c.options.ConfigPath = tempCfg
|
||||
c.options.NodeSecret = ""
|
||||
c.options.PortalUrl = ""
|
||||
c.options.PortalToken = ""
|
||||
c.options.OptionsYaml = filepath.Join(tempCfg, "options.yml")
|
||||
// Clear values potentially loaded at NewConfig creation.
|
||||
c.options.NodeSecret = ""
|
||||
c.options.PortalUrl = ""
|
||||
c.options.PortalToken = ""
|
||||
c.options.OptionsYaml = filepath.Join(tempCfg, "options.yml")
|
||||
// Clear values that may have been loaded from repo fixtures before we
|
||||
// isolated the config path.
|
||||
c.options.NodeSecret = ""
|
||||
c.options.PortalUrl = ""
|
||||
c.options.PortalToken = ""
|
||||
c.options.OptionsYaml = filepath.Join(tempCfg, "options.yml")
|
||||
|
||||
// PortalConfigPath always points to a "cluster" subfolder under ConfigPath.
|
||||
expectedCluster := filepath.Join(c.ConfigPath(), fs.ClusterDir)
|
||||
@@ -54,9 +69,14 @@ func TestConfig_Cluster(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("PortalAndSecrets", func(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
// Isolate config so defaults aren't overridden by repo fixtures: set config-path
|
||||
// before creating the Config so NewConfig does not load repository options.yml.
|
||||
tempCfg := t.TempDir()
|
||||
ctx := CliTestContext()
|
||||
assert.NoError(t, ctx.Set("config-path", tempCfg))
|
||||
c := NewConfig(ctx)
|
||||
|
||||
// Defaults
|
||||
// Defaults (no options.yml present)
|
||||
assert.Equal(t, "", c.PortalUrl())
|
||||
assert.Equal(t, "", c.PortalToken())
|
||||
assert.Equal(t, "", c.NodeSecret())
|
||||
|
||||
@@ -9,14 +9,21 @@ import (
|
||||
|
||||
func TestConfig_DatabaseDriver(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
// Ensure defaults not overridden by repo fixtures.
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
c.options.DatabaseServer = ""
|
||||
c.options.DatabaseName = ""
|
||||
c.options.DatabaseUser = ""
|
||||
c.options.DatabasePassword = ""
|
||||
driver := c.DatabaseDriver()
|
||||
assert.Equal(t, SQLite3, driver)
|
||||
}
|
||||
|
||||
func TestConfig_DatabaseDriverName(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
driver := c.DatabaseDriverName()
|
||||
assert.Equal(t, "SQLite", driver)
|
||||
}
|
||||
@@ -68,7 +75,8 @@ func TestConfig_ParseDatabaseDsn(t *testing.T) {
|
||||
|
||||
func TestConfig_DatabaseServer(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, "", c.DatabaseServer())
|
||||
c.options.DatabaseServer = "test"
|
||||
assert.Equal(t, "", c.DatabaseServer())
|
||||
@@ -76,37 +84,43 @@ func TestConfig_DatabaseServer(t *testing.T) {
|
||||
|
||||
func TestConfig_DatabaseHost(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, "", c.DatabaseHost())
|
||||
}
|
||||
|
||||
func TestConfig_DatabasePort(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, 0, c.DatabasePort())
|
||||
}
|
||||
|
||||
func TestConfig_DatabasePortString(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, "", c.DatabasePortString())
|
||||
}
|
||||
|
||||
func TestConfig_DatabaseName(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, "/go/src/github.com/photoprism/photoprism/storage/testdata/index.db?_busy_timeout=5000", c.DatabaseName())
|
||||
}
|
||||
|
||||
func TestConfig_DatabaseUser(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, "", c.DatabaseUser())
|
||||
}
|
||||
|
||||
func TestConfig_DatabasePassword(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
assert.Equal(t, "", c.DatabasePassword())
|
||||
|
||||
// Test setting the password via secret file.
|
||||
@@ -122,7 +136,8 @@ func TestConfig_DatabasePassword(t *testing.T) {
|
||||
|
||||
func TestConfig_DatabaseDsn(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
driver := c.DatabaseDriver()
|
||||
assert.Equal(t, SQLite3, driver)
|
||||
c.options.DatabaseDsn = ""
|
||||
@@ -140,7 +155,13 @@ func TestConfig_DatabaseDsn(t *testing.T) {
|
||||
|
||||
func TestConfig_DatabaseFile(t *testing.T) {
|
||||
c := NewConfig(CliTestContext())
|
||||
|
||||
// Ensure SQLite defaults
|
||||
c.options.DatabaseDriver = ""
|
||||
c.options.DatabaseDsn = ""
|
||||
c.options.DatabaseServer = ""
|
||||
c.options.DatabaseName = ""
|
||||
c.options.DatabaseUser = ""
|
||||
c.options.DatabasePassword = ""
|
||||
driver := c.DatabaseDriver()
|
||||
assert.Equal(t, SQLite3, driver)
|
||||
c.options.DatabaseDsn = ""
|
||||
|
||||
@@ -12,8 +12,16 @@ var (
|
||||
extInit sync.Once
|
||||
extMutex sync.Mutex
|
||||
extensions atomic.Value
|
||||
// Early extension registry for hooks that must run before DB connect.
|
||||
earlyExtInit sync.Once
|
||||
earlyExtMutex sync.Mutex
|
||||
earlyExtensions atomic.Value
|
||||
)
|
||||
|
||||
// TODO: Provide a test-only reset for earlyExtensions and extensions if we ever
|
||||
// need to reinitialize different early hooks across multiple test packages.
|
||||
// sync.Once currently prevents re-running initializers within the same process.
|
||||
|
||||
// Register registers a new package extension.
|
||||
func Register(name string, initConfig func(c *Config) error, clientConfig func(c *Config, t ClientType) Map) {
|
||||
extMutex.Lock()
|
||||
@@ -23,6 +31,17 @@ func Register(name string, initConfig func(c *Config) error, clientConfig func(c
|
||||
extensions.Store(append(n, Extension{name: name, init: initConfig, clientValues: clientConfig}))
|
||||
}
|
||||
|
||||
// RegisterEarly registers a package extension that should run before the
|
||||
// database connection is established. Use this for hooks that may influence
|
||||
// DB settings or other early configuration.
|
||||
func RegisterEarly(name string, initConfig func(c *Config) error, clientConfig func(c *Config, t ClientType) Map) {
|
||||
earlyExtMutex.Lock()
|
||||
defer earlyExtMutex.Unlock()
|
||||
|
||||
n, _ := earlyExtensions.Load().(Extensions)
|
||||
earlyExtensions.Store(append(n, Extension{name: name, init: initConfig, clientValues: clientConfig}))
|
||||
}
|
||||
|
||||
// Ext returns all registered package extensions.
|
||||
func Ext() (ext Extensions) {
|
||||
extMutex.Lock()
|
||||
@@ -33,6 +52,16 @@ func Ext() (ext Extensions) {
|
||||
return ext
|
||||
}
|
||||
|
||||
// EarlyExt returns all registered early package extensions.
|
||||
func EarlyExt() (ext Extensions) {
|
||||
earlyExtMutex.Lock()
|
||||
defer earlyExtMutex.Unlock()
|
||||
|
||||
ext, _ = earlyExtensions.Load().(Extensions)
|
||||
|
||||
return ext
|
||||
}
|
||||
|
||||
// Extensions represents a list of package extensions.
|
||||
type Extensions []Extension
|
||||
|
||||
@@ -57,3 +86,18 @@ func (ext Extensions) Init(c *Config) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// InitEarly initializes early registered extensions.
|
||||
func (ext Extensions) InitEarly(c *Config) {
|
||||
earlyExtInit.Do(func() {
|
||||
for _, e := range ext {
|
||||
start := time.Now()
|
||||
|
||||
if err := e.init(c); err != nil {
|
||||
log.Warnf("config: %s when loading early %s extension", err, clean.Log(e.name))
|
||||
} else {
|
||||
log.Tracef("config: early %s extension loaded [%s]", clean.Log(e.name), time.Since(start))
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
336
internal/service/cluster/instance/bootstrap.go
Normal file
336
internal/service/cluster/instance/bootstrap.go
Normal file
@@ -0,0 +1,336 @@
|
||||
package instance
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
yaml "gopkg.in/yaml.v2"
|
||||
|
||||
"github.com/photoprism/photoprism/internal/config"
|
||||
"github.com/photoprism/photoprism/internal/event"
|
||||
"github.com/photoprism/photoprism/internal/service/cluster"
|
||||
"github.com/photoprism/photoprism/pkg/clean"
|
||||
"github.com/photoprism/photoprism/pkg/fs"
|
||||
)
|
||||
|
||||
var log = event.Log
|
||||
|
||||
func init() {
|
||||
// Register early so this can adjust DB settings before connectDb().
|
||||
config.RegisterEarly("cluster-instance", InitConfig, nil)
|
||||
}
|
||||
|
||||
// InitConfig performs instance bootstrap: optional registration with the Portal
|
||||
// and theme installation. Runs early during config.Init().
|
||||
func InitConfig(c *config.Config) error {
|
||||
if !cluster.BootstrapAutoJoinEnabled && !cluster.BootstrapAutoThemeEnabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Skip on portal nodes and unknown node types.
|
||||
if c.IsPortal() || c.NodeType() != cluster.Instance {
|
||||
return nil
|
||||
}
|
||||
|
||||
portalURL := strings.TrimSpace(c.PortalUrl())
|
||||
portalToken := strings.TrimSpace(c.PortalToken())
|
||||
if portalURL == "" || portalToken == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
u, err := url.Parse(portalURL)
|
||||
if err != nil || u.Scheme == "" || u.Host == "" {
|
||||
log.Warnf("cluster: invalid portal url %s", clean.Log(portalURL))
|
||||
return nil
|
||||
}
|
||||
|
||||
// Enforce TLS for non-local URLs.
|
||||
if u.Scheme != "https" && !isLocalHost(u.Hostname()) {
|
||||
log.Warnf("cluster: refusing non-TLS portal url %s on non-local host", clean.Log(portalURL))
|
||||
return nil
|
||||
}
|
||||
|
||||
// Register with retry policy.
|
||||
if cluster.BootstrapAutoJoinEnabled {
|
||||
if err := registerWithPortal(c, u, portalToken); err != nil {
|
||||
// Registration errors are expected when the Portal is temporarily unavailable
|
||||
// or not configured with cluster endpoints (404). Keep as warn to signal
|
||||
// exhaustion/terminal errors; per-attempt details are logged at debug level.
|
||||
log.Warnf("cluster: register failed (%s)", clean.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// Pull theme if missing.
|
||||
if cluster.BootstrapAutoThemeEnabled {
|
||||
if err := installThemeIfMissing(c, u, portalToken); err != nil {
|
||||
// Theme install failures are non-critical; log at debug to avoid noise.
|
||||
log.Debugf("cluster: theme install skipped/failed (%s)", clean.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func isLocalHost(h string) bool {
|
||||
switch strings.ToLower(h) {
|
||||
case "localhost", "127.0.0.1", "::1":
|
||||
return true
|
||||
default:
|
||||
// TODO: Consider treating RFC1918/link-local hosts as local for TLS enforcement
|
||||
// if the operator explicitly opts in (e.g., via a policy var). Keep simple for now.
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func newHTTPClient(timeout time.Duration) *http.Client {
|
||||
// TODO: Consider reusing a shared *http.Transport with sane defaults and enabling
|
||||
// proxy support explicitly if required. For now, rely on net/http defaults and
|
||||
// the HTTPS_PROXY set in config.Init().
|
||||
return &http.Client{Timeout: timeout}
|
||||
}
|
||||
|
||||
func registerWithPortal(c *config.Config, portal *url.URL, token string) error {
|
||||
maxAttempts := cluster.BootstrapRegisterMaxAttempts
|
||||
delay := cluster.BootstrapRegisterRetryDelay
|
||||
timeout := cluster.BootstrapRegisterTimeout
|
||||
|
||||
endpoint := *portal
|
||||
endpoint.Path = strings.TrimRight(endpoint.Path, "/") + "/api/v1/cluster/nodes/register"
|
||||
|
||||
// Decide if DB rotation is desired as per spec: only if driver is MySQL/MariaDB
|
||||
// and no DSN/fields are set (raw options) and no password is provided via file.
|
||||
opts := c.Options()
|
||||
driver := c.DatabaseDriver()
|
||||
wantRotateDB := (driver == config.MySQL || driver == config.MariaDB) &&
|
||||
opts.DatabaseDsn == "" && opts.DatabaseName == "" && opts.DatabaseUser == "" && opts.DatabasePassword == "" &&
|
||||
c.DatabasePassword() == ""
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"nodeName": c.NodeName(),
|
||||
"nodeType": string(cluster.Instance), // JSON wire format is string
|
||||
"internalUrl": c.InternalUrl(),
|
||||
}
|
||||
if wantRotateDB {
|
||||
payload["rotate"] = true
|
||||
}
|
||||
|
||||
bodyBytes, _ := json.Marshal(payload)
|
||||
|
||||
for attempt := 1; attempt <= maxAttempts; attempt++ {
|
||||
req, _ := http.NewRequest(http.MethodPost, endpoint.String(), strings.NewReader(string(bodyBytes)))
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Accept", "application/json")
|
||||
|
||||
resp, err := newHTTPClient(timeout).Do(req)
|
||||
if err != nil {
|
||||
if attempt < maxAttempts {
|
||||
log.Debugf("cluster: register attempt %d/%d error: %s", attempt, maxAttempts, clean.Error(err))
|
||||
time.Sleep(delay)
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Ensure body is closed after handling the response.
|
||||
defer resp.Body.Close()
|
||||
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK, http.StatusCreated:
|
||||
var r cluster.RegisterResponse
|
||||
dec := json.NewDecoder(resp.Body)
|
||||
if err := dec.Decode(&r); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := persistRegistration(c, &r, wantRotateDB); err != nil {
|
||||
return err
|
||||
}
|
||||
if resp.StatusCode == http.StatusCreated {
|
||||
log.Infof("cluster: registered as %s (%d)", clean.LogQuote(r.Node.Name), resp.StatusCode)
|
||||
} else {
|
||||
log.Infof("cluster: registration ok (%d)", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
case http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound:
|
||||
// Terminal errors (no retry). 404 likely indicates a Portal without cluster endpoints.
|
||||
return errors.New(resp.Status)
|
||||
case http.StatusTooManyRequests:
|
||||
if attempt < maxAttempts {
|
||||
log.Debugf("cluster: register attempt %d/%d rate limited", attempt, maxAttempts)
|
||||
time.Sleep(delay)
|
||||
continue
|
||||
}
|
||||
return errors.New(resp.Status)
|
||||
case http.StatusConflict, http.StatusBadRequest:
|
||||
// Do not retry on 400/409 per spec intent.
|
||||
return errors.New(resp.Status)
|
||||
default:
|
||||
if attempt < maxAttempts {
|
||||
log.Debugf("cluster: register attempt %d/%d server responded %s", attempt, maxAttempts, resp.Status)
|
||||
// TODO: Consider exponential backoff with jitter instead of constant delay.
|
||||
time.Sleep(delay)
|
||||
continue
|
||||
}
|
||||
return errors.New(resp.Status)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isTemporary(err error) bool {
|
||||
var nerr net.Error
|
||||
return errors.As(err, &nerr) && nerr.Timeout()
|
||||
}
|
||||
|
||||
func persistRegistration(c *config.Config, r *cluster.RegisterResponse, wantRotateDB bool) error {
|
||||
updates := map[string]interface{}{}
|
||||
|
||||
// Persist node secret only if missing locally and provided by server.
|
||||
if r.Secrets != nil && r.Secrets.NodeSecret != "" && c.NodeSecret() == "" {
|
||||
updates["NodeSecret"] = r.Secrets.NodeSecret
|
||||
}
|
||||
|
||||
// Persist DB settings only if rotation was requested and driver is MySQL/MariaDB
|
||||
// and local DB not configured (as checked before calling).
|
||||
if wantRotateDB {
|
||||
if r.DB.DSN != "" {
|
||||
updates["DatabaseDriver"] = config.MySQL
|
||||
updates["DatabaseDsn"] = r.DB.DSN
|
||||
} else if r.DB.Name != "" && r.DB.User != "" && r.DB.Password != "" {
|
||||
server := r.DB.Host
|
||||
if r.DB.Port > 0 {
|
||||
server = net.JoinHostPort(r.DB.Host, strconv.Itoa(r.DB.Port))
|
||||
}
|
||||
updates["DatabaseDriver"] = config.MySQL
|
||||
updates["DatabaseServer"] = server
|
||||
updates["DatabaseName"] = r.DB.Name
|
||||
updates["DatabaseUser"] = r.DB.User
|
||||
updates["DatabasePassword"] = r.DB.Password
|
||||
}
|
||||
}
|
||||
|
||||
if len(updates) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := mergeOptionsYaml(c, updates); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Reload into memory so later code paths see updated values during this run.
|
||||
_ = c.Options().Load(c.OptionsYaml())
|
||||
|
||||
if hasDBUpdate(updates) {
|
||||
log.Infof("cluster: database settings applied; restart required to take effect")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func hasDBUpdate(m map[string]interface{}) bool {
|
||||
if _, ok := m["DatabaseDsn"]; ok {
|
||||
return true
|
||||
}
|
||||
if _, ok := m["DatabaseName"]; ok {
|
||||
return true
|
||||
}
|
||||
if _, ok := m["DatabaseUser"]; ok {
|
||||
return true
|
||||
}
|
||||
if _, ok := m["DatabasePassword"]; ok {
|
||||
return true
|
||||
}
|
||||
if _, ok := m["DatabaseServer"]; ok {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func mergeOptionsYaml(c *config.Config, updates map[string]interface{}) error {
|
||||
if err := fs.MkdirAll(c.ConfigPath()); err != nil {
|
||||
return err
|
||||
}
|
||||
fileName := c.OptionsYaml()
|
||||
|
||||
var m map[string]interface{}
|
||||
if fs.FileExists(fileName) {
|
||||
if b, err := os.ReadFile(fileName); err == nil && len(b) > 0 {
|
||||
_ = yaml.Unmarshal(b, &m)
|
||||
}
|
||||
}
|
||||
if m == nil {
|
||||
m = map[string]interface{}{}
|
||||
}
|
||||
for k, v := range updates {
|
||||
m[k] = v
|
||||
}
|
||||
|
||||
b, err := yaml.Marshal(m)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(fileName, b, 0o644)
|
||||
}
|
||||
|
||||
// installThemeIfMissing downloads and installs the Portal-provided theme if the
|
||||
// local theme directory is missing or lacks an app.js file.
|
||||
func installThemeIfMissing(c *config.Config, portal *url.URL, token string) error {
|
||||
themeDir := c.ThemePath()
|
||||
need := !fs.PathExists(themeDir) || (cluster.BootstrapThemeInstallOnlyIfMissingJS && !fs.FileExists(filepath.Join(themeDir, "app.js")))
|
||||
if !need && !cluster.BootstrapAllowThemeOverwrite {
|
||||
return nil
|
||||
}
|
||||
|
||||
endpoint := *portal
|
||||
endpoint.Path = strings.TrimRight(endpoint.Path, "/") + "/api/v1/cluster/theme"
|
||||
|
||||
req, _ := http.NewRequest(http.MethodGet, endpoint.String(), nil)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
req.Header.Set("Accept", "application/zip")
|
||||
|
||||
resp, err := newHTTPClient(cluster.BootstrapRegisterTimeout).Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK:
|
||||
// Save to temp zip.
|
||||
if err := fs.MkdirAll(c.TempPath()); err != nil {
|
||||
return err
|
||||
}
|
||||
zipName := filepath.Join(c.TempPath(), "cluster-theme.zip")
|
||||
out, err := os.Create(zipName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = io.Copy(out, resp.Body); err != nil {
|
||||
_ = out.Close()
|
||||
return err
|
||||
}
|
||||
_ = out.Close()
|
||||
|
||||
// Extract with moderate limits.
|
||||
if err := fs.MkdirAll(themeDir); err != nil {
|
||||
return err
|
||||
}
|
||||
_, _, unzipErr := fs.Unzip(zipName, themeDir, 32*fs.MB, 512*fs.MB)
|
||||
return unzipErr
|
||||
case http.StatusNotFound:
|
||||
// No theme configured at Portal.
|
||||
return nil
|
||||
case http.StatusUnauthorized, http.StatusForbidden:
|
||||
return errors.New(resp.Status)
|
||||
default:
|
||||
return errors.New(resp.Status)
|
||||
}
|
||||
}
|
||||
212
internal/service/cluster/instance/bootstrap_test.go
Normal file
212
internal/service/cluster/instance/bootstrap_test.go
Normal file
@@ -0,0 +1,212 @@
|
||||
package instance
|
||||
|
||||
import (
|
||||
"archive/zip"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"github.com/photoprism/photoprism/internal/config"
|
||||
"github.com/photoprism/photoprism/internal/service/cluster"
|
||||
)
|
||||
|
||||
func TestInitConfig_NoPortal_NoOp(t *testing.T) {
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", t.TempDir())
|
||||
c := config.NewTestConfig("bootstrap-np")
|
||||
// Default NodeType() resolves to instance; no Portal configured.
|
||||
assert.Equal(t, cluster.Instance, c.NodeType())
|
||||
assert.NoError(t, InitConfig(c))
|
||||
}
|
||||
|
||||
func TestRegister_PersistSecretAndDB(t *testing.T) {
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", t.TempDir())
|
||||
// Fake Portal server.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api/v1/cluster/nodes/register":
|
||||
// Minimal successful registration with secrets + DSN.
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
resp := cluster.RegisterResponse{
|
||||
Node: cluster.Node{Name: "pp-node-01"},
|
||||
Secrets: &cluster.RegisterSecrets{NodeSecret: "SECRET"},
|
||||
DB: cluster.RegisterDB{Host: "db.local", Port: 3306, Name: "pp_db", User: "pp_user", Password: "pp_pw", DSN: "pp_user:pp_pw@tcp(db.local:3306)/pp_db?charset=utf8mb4&parseTime=true"},
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(resp)
|
||||
case "/api/v1/cluster/theme":
|
||||
// No theme for this test.
|
||||
http.NotFound(w, r)
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := config.NewTestConfig("bootstrap-reg")
|
||||
// Configure Portal.
|
||||
c.Options().PortalUrl = srv.URL
|
||||
c.Options().PortalToken = "t0k3n"
|
||||
// Gate rotate=true: driver mysql and no DSN/fields.
|
||||
c.Options().DatabaseDriver = config.MySQL
|
||||
c.Options().DatabaseDsn = ""
|
||||
c.Options().DatabaseName = ""
|
||||
c.Options().DatabaseUser = ""
|
||||
c.Options().DatabasePassword = ""
|
||||
|
||||
// Run bootstrap.
|
||||
assert.NoError(t, InitConfig(c))
|
||||
|
||||
// Options should be reloaded; check values.
|
||||
assert.Equal(t, "SECRET", c.NodeSecret())
|
||||
// DSN branch should be preferred and persisted.
|
||||
assert.Contains(t, c.Options().DatabaseDsn, "@tcp(db.local:3306)/pp_db")
|
||||
assert.Equal(t, config.MySQL, c.Options().DatabaseDriver)
|
||||
}
|
||||
|
||||
func TestThemeInstall_Missing(t *testing.T) {
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", t.TempDir())
|
||||
// Build a tiny zip in-memory with one file style.css
|
||||
var buf bytes.Buffer
|
||||
zw := zip.NewWriter(&buf)
|
||||
f, _ := zw.Create("style.css")
|
||||
_, _ = f.Write([]byte("body{}\n"))
|
||||
_ = zw.Close()
|
||||
|
||||
// Fake Portal server.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api/v1/cluster/nodes/register":
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(cluster.RegisterResponse{Node: cluster.Node{Name: "pp-node-01"}})
|
||||
case "/api/v1/cluster/theme":
|
||||
w.Header().Set("Content-Type", "application/zip")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(buf.Bytes())
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := config.NewTestConfig("bootstrap-theme")
|
||||
// Point Portal.
|
||||
c.Options().PortalUrl = srv.URL
|
||||
c.Options().PortalToken = "t0k3n"
|
||||
|
||||
// Ensure theme dir is empty and unique.
|
||||
tempTheme, err := os.MkdirTemp("", "pp-theme-*")
|
||||
assert.NoError(t, err)
|
||||
defer func() { _ = os.RemoveAll(tempTheme) }()
|
||||
c.SetThemePath(tempTheme)
|
||||
// Remove style.css if any left from previous runs.
|
||||
_ = os.Remove(filepath.Join(tempTheme, "style.css"))
|
||||
|
||||
// Run bootstrap.
|
||||
assert.NoError(t, InitConfig(c))
|
||||
|
||||
// Expect style.css to exist in theme dir.
|
||||
_, statErr := os.Stat(filepath.Join(tempTheme, "style.css"))
|
||||
assert.NoError(t, statErr)
|
||||
}
|
||||
|
||||
func TestRegister_SQLite_NoDBPersist(t *testing.T) {
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", t.TempDir())
|
||||
// Portal responds with DB DSN, but local driver is SQLite → must not persist DB.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api/v1/cluster/nodes/register":
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
resp := cluster.RegisterResponse{
|
||||
Node: cluster.Node{Name: "pp-node-01"},
|
||||
Secrets: &cluster.RegisterSecrets{NodeSecret: "SECRET"},
|
||||
DB: cluster.RegisterDB{Host: "db.local", Port: 3306, Name: "pp_db", User: "pp_user", Password: "pp_pw", DSN: "pp_user:pp_pw@tcp(db.local:3306)/pp_db?charset=utf8mb4&parseTime=true"},
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(resp)
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := config.NewTestConfig("bootstrap-sqlite")
|
||||
// SQLite driver by default; set Portal.
|
||||
c.Options().PortalUrl = srv.URL
|
||||
c.Options().PortalToken = "t0k3n"
|
||||
// Remember original DSN so we can ensure it is not changed.
|
||||
origDSN := c.Options().DatabaseDsn
|
||||
t.Cleanup(func() { _ = os.Remove(origDSN) })
|
||||
|
||||
// Run bootstrap.
|
||||
assert.NoError(t, InitConfig(c))
|
||||
|
||||
// NodeSecret should persist, but DB should remain SQLite (no DSN update).
|
||||
assert.Equal(t, "SECRET", c.NodeSecret())
|
||||
assert.Equal(t, config.SQLite3, c.DatabaseDriver())
|
||||
assert.Equal(t, origDSN, c.Options().DatabaseDsn)
|
||||
}
|
||||
|
||||
func TestRegister_404_NoRetry(t *testing.T) {
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", t.TempDir())
|
||||
var hits int
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/api/v1/cluster/nodes/register" {
|
||||
hits++
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
http.NotFound(w, r)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := config.NewTestConfig("bootstrap-404")
|
||||
c.Options().PortalUrl = srv.URL
|
||||
c.Options().PortalToken = "t0k3n"
|
||||
|
||||
// Run bootstrap; registration should attempt once and stop on 404.
|
||||
_ = InitConfig(c)
|
||||
|
||||
assert.Equal(t, 1, hits)
|
||||
}
|
||||
|
||||
func TestThemeInstall_SkipWhenAppJsExists(t *testing.T) {
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", t.TempDir())
|
||||
// Portal returns a valid zip, but theme dir already has app.js → skip.
|
||||
var served int
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/api/v1/cluster/theme" {
|
||||
served++
|
||||
w.Header().Set("Content-Type", "application/zip")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
zw := zip.NewWriter(w)
|
||||
_, _ = zw.Create("style.css")
|
||||
_ = zw.Close()
|
||||
return
|
||||
}
|
||||
http.NotFound(w, r)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := config.NewTestConfig("bootstrap-theme-skip")
|
||||
c.Options().PortalUrl = srv.URL
|
||||
c.Options().PortalToken = "t0k3n"
|
||||
|
||||
// Prepare theme dir with app.js
|
||||
tempTheme, err := os.MkdirTemp("", "pp-theme-*")
|
||||
assert.NoError(t, err)
|
||||
defer func() { _ = os.RemoveAll(tempTheme) }()
|
||||
c.SetThemePath(tempTheme)
|
||||
assert.NoError(t, os.WriteFile(filepath.Join(tempTheme, "app.js"), []byte("// app\n"), 0o644))
|
||||
|
||||
assert.NoError(t, InitConfig(c))
|
||||
// Should have skipped request because app.js already exists.
|
||||
assert.Equal(t, 0, served)
|
||||
_, statErr := os.Stat(filepath.Join(tempTheme, "style.css"))
|
||||
assert.Error(t, statErr)
|
||||
}
|
||||
36
internal/service/cluster/policy.go
Normal file
36
internal/service/cluster/policy.go
Normal file
@@ -0,0 +1,36 @@
|
||||
package cluster
|
||||
|
||||
import "time"
|
||||
|
||||
// BootstrapAutoJoinEnabled indicates whether cluster bootstrap logic is enabled
|
||||
// for nodes by default. Portal nodes ignore this value; gating is decided by
|
||||
// runtime checks (e.g., conf.IsPortal() and conf.NodeType()).
|
||||
var BootstrapAutoJoinEnabled = true
|
||||
|
||||
// BootstrapAutoThemeEnabled indicates whether bootstrap should attempt to
|
||||
// download and install a Portal-provided theme when appropriate.
|
||||
var BootstrapAutoThemeEnabled = true
|
||||
|
||||
// BootstrapRegisterMaxAttempts defines how many attempts the bootstrap logic
|
||||
// makes when contacting the Portal for registration before giving up.
|
||||
var BootstrapRegisterMaxAttempts = 6
|
||||
|
||||
// BootstrapRegisterRetryDelay defines the delay between registration attempts
|
||||
// when the Portal is temporarily unavailable.
|
||||
var BootstrapRegisterRetryDelay = 15 * time.Second
|
||||
|
||||
// BootstrapRegisterTimeout defines the HTTP client timeout per registration
|
||||
// request to the Portal.
|
||||
var BootstrapRegisterTimeout = 15 * time.Second
|
||||
|
||||
// BootstrapThemeInstallOnlyIfMissingJS ensures theme installation only happens
|
||||
// when the local theme directory is missing or does not contain an app.js file.
|
||||
var BootstrapThemeInstallOnlyIfMissingJS = true
|
||||
|
||||
// BootstrapAllowThemeOverwrite indicates whether bootstrap may overwrite an
|
||||
// existing local theme. The default is false to protect local modifications.
|
||||
var BootstrapAllowThemeOverwrite = false
|
||||
|
||||
// TODO: Consider allowing these policy defaults to be overridden via environment
|
||||
// variables (e.g., for CI) without exposing user-facing config flags. Keep the
|
||||
// public surface area small until we see demand.
|
||||
@@ -1,62 +1,62 @@
|
||||
package registry
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
yaml "gopkg.in/yaml.v2"
|
||||
yaml "gopkg.in/yaml.v2"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"github.com/photoprism/photoprism/internal/config"
|
||||
"github.com/photoprism/photoprism/internal/config"
|
||||
)
|
||||
|
||||
// TestFindByNameDeterministic verifies that FindByName returns the most
|
||||
// recently updated node when multiple registry entries share the same Name.
|
||||
func TestFindByNameDeterministic(t *testing.T) {
|
||||
// Isolate storage/config to avoid interference from other tests.
|
||||
tmp := t.TempDir()
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", tmp)
|
||||
// Isolate storage/config to avoid interference from other tests.
|
||||
tmp := t.TempDir()
|
||||
t.Setenv("PHOTOPRISM_STORAGE_PATH", tmp)
|
||||
|
||||
conf := config.NewTestConfig("cluster-registry-findbyname")
|
||||
conf := config.NewTestConfig("cluster-registry-findbyname")
|
||||
|
||||
r, err := NewFileRegistry(conf)
|
||||
assert.NoError(t, err)
|
||||
r, err := NewFileRegistry(conf)
|
||||
assert.NoError(t, err)
|
||||
|
||||
// Two nodes with the same name but different UpdatedAt timestamps.
|
||||
old := Node{
|
||||
ID: "id-old",
|
||||
Name: "pp-node-01",
|
||||
Type: "instance",
|
||||
CreatedAt: "2024-01-01T00:00:00Z",
|
||||
UpdatedAt: "2024-01-01T00:00:00Z",
|
||||
}
|
||||
newer := Node{
|
||||
ID: "id-new",
|
||||
Name: "pp-node-01",
|
||||
Type: "instance",
|
||||
CreatedAt: "2024-02-01T00:00:00Z",
|
||||
UpdatedAt: "2024-02-01T00:00:00Z",
|
||||
}
|
||||
// Two nodes with the same name but different UpdatedAt timestamps.
|
||||
old := Node{
|
||||
ID: "id-old",
|
||||
Name: "pp-node-01",
|
||||
Type: "instance",
|
||||
CreatedAt: "2024-01-01T00:00:00Z",
|
||||
UpdatedAt: "2024-01-01T00:00:00Z",
|
||||
}
|
||||
newer := Node{
|
||||
ID: "id-new",
|
||||
Name: "pp-node-01",
|
||||
Type: "instance",
|
||||
CreatedAt: "2024-02-01T00:00:00Z",
|
||||
UpdatedAt: "2024-02-01T00:00:00Z",
|
||||
}
|
||||
|
||||
// Write YAML files directly to avoid timing flakiness.
|
||||
b1, err := yaml.Marshal(old)
|
||||
assert.NoError(t, err)
|
||||
assert.NoError(t, os.WriteFile(r.fileName(old.ID), b1, 0o600))
|
||||
// Write YAML files directly to avoid timing flakiness.
|
||||
b1, err := yaml.Marshal(old)
|
||||
assert.NoError(t, err)
|
||||
assert.NoError(t, os.WriteFile(r.fileName(old.ID), b1, 0o600))
|
||||
|
||||
b2, err := yaml.Marshal(newer)
|
||||
assert.NoError(t, err)
|
||||
assert.NoError(t, os.WriteFile(r.fileName(newer.ID), b2, 0o600))
|
||||
b2, err := yaml.Marshal(newer)
|
||||
assert.NoError(t, err)
|
||||
assert.NoError(t, os.WriteFile(r.fileName(newer.ID), b2, 0o600))
|
||||
|
||||
// Expect the most recently updated node (id-new).
|
||||
got, err := r.FindByName("pp-node-01")
|
||||
assert.NoError(t, err)
|
||||
if assert.NotNil(t, got) {
|
||||
assert.Equal(t, "id-new", got.ID)
|
||||
assert.Equal(t, "pp-node-01", got.Name)
|
||||
}
|
||||
// Expect the most recently updated node (id-new).
|
||||
got, err := r.FindByName("pp-node-01")
|
||||
assert.NoError(t, err)
|
||||
if assert.NotNil(t, got) {
|
||||
assert.Equal(t, "id-new", got.ID)
|
||||
assert.Equal(t, "pp-node-01", got.Name)
|
||||
}
|
||||
|
||||
// Non-existent name should return os.ErrNotExist.
|
||||
_, err = r.FindByName("does-not-exist")
|
||||
assert.ErrorIs(t, err, os.ErrNotExist)
|
||||
// Non-existent name should return os.ErrNotExist.
|
||||
_, err = r.FindByName("does-not-exist")
|
||||
assert.ErrorIs(t, err, os.ErrNotExist)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user