ensure commands for all services
This commit is contained in:
@@ -1,267 +0,0 @@
|
||||
package command
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"path"
|
||||
|
||||
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
|
||||
"github.com/cs3org/reva/v2/cmd/revad/runtime"
|
||||
"github.com/gofrs/uuid"
|
||||
"github.com/oklog/run"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config/parser"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/server/debug"
|
||||
ociscfg "github.com/owncloud/ocis/ocis-pkg/config"
|
||||
"github.com/owncloud/ocis/ocis-pkg/log"
|
||||
"github.com/owncloud/ocis/ocis-pkg/service/external"
|
||||
"github.com/owncloud/ocis/ocis-pkg/sync"
|
||||
"github.com/owncloud/ocis/ocis-pkg/tracing"
|
||||
"github.com/owncloud/ocis/ocis-pkg/version"
|
||||
"github.com/thejerf/suture/v4"
|
||||
"github.com/urfave/cli/v2"
|
||||
)
|
||||
|
||||
// StorageMetadata the entrypoint for the storage-storage-metadata command.
|
||||
//
|
||||
// It provides a ocis-specific storage store metadata (shares,account,settings...)
|
||||
func StorageMetadata(cfg *config.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "storage-metadata",
|
||||
Usage: "start storage-metadata service",
|
||||
Category: "extensions",
|
||||
Before: func(ctx *cli.Context) error {
|
||||
err := parser.ParseConfig(cfg)
|
||||
if err != nil {
|
||||
fmt.Printf("%v", err)
|
||||
}
|
||||
return err
|
||||
},
|
||||
Action: func(c *cli.Context) error {
|
||||
logCfg := cfg.Logging
|
||||
logger := log.NewLogger(
|
||||
log.Level(logCfg.Level),
|
||||
log.File(logCfg.File),
|
||||
log.Pretty(logCfg.Pretty),
|
||||
log.Color(logCfg.Color),
|
||||
)
|
||||
tracing.Configure(cfg.Tracing.Enabled, cfg.Tracing.Type, logger)
|
||||
|
||||
gr := run.Group{}
|
||||
ctx, cancel := func() (context.Context, context.CancelFunc) {
|
||||
if cfg.Context == nil {
|
||||
return context.WithCancel(context.Background())
|
||||
}
|
||||
return context.WithCancel(cfg.Context)
|
||||
}()
|
||||
|
||||
defer cancel()
|
||||
|
||||
pidFile := path.Join(os.TempDir(), "revad-"+c.Command.Name+"-"+uuid.Must(uuid.NewV4()).String()+".pid")
|
||||
rcfg := storageMetadataFromStruct(c, cfg)
|
||||
|
||||
gr.Add(func() error {
|
||||
runtime.RunWithOptions(
|
||||
rcfg,
|
||||
pidFile,
|
||||
runtime.WithLogger(&logger.Logger),
|
||||
)
|
||||
return nil
|
||||
}, func(_ error) {
|
||||
logger.Info().
|
||||
Str("server", c.Command.Name).
|
||||
Msg("Shutting down server")
|
||||
|
||||
cancel()
|
||||
})
|
||||
|
||||
debugServer, err := debug.Server(
|
||||
debug.Logger(logger),
|
||||
debug.Context(ctx),
|
||||
debug.Config(cfg),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
logger.Info().
|
||||
Err(err).
|
||||
Str("server", c.Command.Name+"-debug").
|
||||
Msg("Failed to initialize server")
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
gr.Add(func() error {
|
||||
return debugServer.ListenAndServe()
|
||||
}, func(_ error) {
|
||||
_ = debugServer.Shutdown(ctx)
|
||||
cancel()
|
||||
})
|
||||
|
||||
if !cfg.Supervised {
|
||||
sync.Trap(&gr, cancel)
|
||||
}
|
||||
|
||||
if err := external.RegisterGRPCEndpoint(
|
||||
ctx,
|
||||
"com.owncloud.storage.metadata",
|
||||
uuid.Must(uuid.NewV4()).String(),
|
||||
cfg.GRPC.Addr,
|
||||
version.String,
|
||||
logger,
|
||||
); err != nil {
|
||||
logger.Fatal().Err(err).Msg("failed to register the grpc endpoint")
|
||||
}
|
||||
|
||||
return gr.Run()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// storageMetadataFromStruct will adapt an oCIS config struct into a reva mapstructure to start a reva service.
|
||||
func storageMetadataFromStruct(c *cli.Context, cfg *config.Config) map[string]interface{} {
|
||||
rcfg := map[string]interface{}{
|
||||
"core": map[string]interface{}{
|
||||
"tracing_enabled": cfg.Tracing.Enabled,
|
||||
"tracing_endpoint": cfg.Tracing.Endpoint,
|
||||
"tracing_collector": cfg.Tracing.Collector,
|
||||
"tracing_service_name": c.Command.Name,
|
||||
},
|
||||
"shared": map[string]interface{}{
|
||||
"jwt_secret": cfg.TokenManager.JWTSecret,
|
||||
"gatewaysvc": cfg.Reva.Address,
|
||||
"skip_user_groups_in_token": cfg.SkipUserGroupsInToken,
|
||||
},
|
||||
"grpc": map[string]interface{}{
|
||||
"network": cfg.GRPC.Protocol,
|
||||
"address": cfg.GRPC.Addr,
|
||||
"services": map[string]interface{}{
|
||||
"gateway": map[string]interface{}{
|
||||
// registries are located on the gateway
|
||||
"authregistrysvc": cfg.GRPC.Addr,
|
||||
"storageregistrysvc": cfg.GRPC.Addr,
|
||||
// user metadata is located on the users services
|
||||
"userprovidersvc": cfg.GRPC.Addr,
|
||||
"groupprovidersvc": cfg.GRPC.Addr,
|
||||
"permissionssvc": cfg.GRPC.Addr,
|
||||
// other
|
||||
"disable_home_creation_on_login": true, // metadata manually creates a space
|
||||
// metadata always uses the simple upload, so no transfer secret or datagateway needed
|
||||
},
|
||||
"userprovider": map[string]interface{}{
|
||||
"driver": "memory",
|
||||
"drivers": map[string]interface{}{
|
||||
"memory": map[string]interface{}{
|
||||
"users": map[string]interface{}{
|
||||
"serviceuser": map[string]interface{}{
|
||||
"id": map[string]interface{}{
|
||||
"opaqueId": cfg.MetadataUserID,
|
||||
"idp": "internal",
|
||||
"type": userpb.UserType_USER_TYPE_PRIMARY,
|
||||
},
|
||||
"username": "serviceuser",
|
||||
"display_name": "System User",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"authregistry": map[string]interface{}{
|
||||
"driver": "static",
|
||||
"drivers": map[string]interface{}{
|
||||
"static": map[string]interface{}{
|
||||
"rules": map[string]interface{}{
|
||||
"machine": cfg.GRPC.Addr,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"authprovider": map[string]interface{}{
|
||||
"auth_manager": "machine",
|
||||
"auth_managers": map[string]interface{}{
|
||||
"machine": map[string]interface{}{
|
||||
"api_key": cfg.MachineAuthAPIKey,
|
||||
"gateway_addr": cfg.GRPC.Addr,
|
||||
},
|
||||
},
|
||||
},
|
||||
"permissions": map[string]interface{}{
|
||||
"driver": "demo",
|
||||
"drivers": map[string]interface{}{
|
||||
"demo": map[string]interface{}{},
|
||||
},
|
||||
},
|
||||
"storageregistry": map[string]interface{}{
|
||||
"driver": "static",
|
||||
"drivers": map[string]interface{}{
|
||||
"static": map[string]interface{}{
|
||||
"rules": map[string]interface{}{
|
||||
"/": map[string]interface{}{
|
||||
"address": cfg.GRPC.Addr,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"storageprovider": map[string]interface{}{
|
||||
"driver": cfg.Driver,
|
||||
"drivers": config.MetadataDrivers(cfg),
|
||||
"data_server_url": cfg.DataServerURL,
|
||||
"tmp_folder": cfg.TempFolder,
|
||||
},
|
||||
},
|
||||
},
|
||||
"http": map[string]interface{}{
|
||||
"network": cfg.HTTP.Protocol,
|
||||
"address": cfg.HTTP.Addr,
|
||||
// no datagateway needed as the metadata clients directly talk to the dataprovider with the simple protocol
|
||||
"services": map[string]interface{}{
|
||||
"dataprovider": map[string]interface{}{
|
||||
"prefix": "data",
|
||||
"driver": cfg.Driver,
|
||||
"drivers": config.MetadataDrivers(cfg),
|
||||
"timeout": 86400,
|
||||
"insecure": cfg.DataProviderInsecure,
|
||||
"disable_tus": true,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
return rcfg
|
||||
}
|
||||
|
||||
// SutureService allows for the storage-metadata command to be embedded and supervised by a suture supervisor tree.
|
||||
type MetadataSutureService struct {
|
||||
cfg *config.Config
|
||||
}
|
||||
|
||||
// NewSutureService creates a new storagemetadata.SutureService
|
||||
func NewStorageMetadata(cfg *ociscfg.Config) suture.Service {
|
||||
cfg.StorageMetadata.Commons = cfg.Commons
|
||||
return MetadataSutureService{
|
||||
cfg: cfg.StorageMetadata,
|
||||
}
|
||||
}
|
||||
|
||||
func (s MetadataSutureService) Serve(ctx context.Context) error {
|
||||
s.cfg.Context = ctx
|
||||
f := &flag.FlagSet{}
|
||||
cmdFlags := StorageMetadata(s.cfg).Flags
|
||||
for k := range cmdFlags {
|
||||
if err := cmdFlags[k].Apply(f); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
cliCtx := cli.NewContext(nil, f, nil)
|
||||
if StorageMetadata(s.cfg).Before != nil {
|
||||
if err := StorageMetadata(s.cfg).Before(cliCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := StorageMetadata(s.cfg).Action(cliCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package command
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config/parser"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/logging"
|
||||
"github.com/urfave/cli/v2"
|
||||
)
|
||||
|
||||
// Health is the entrypoint for the health command.
|
||||
func Health(cfg *config.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "health",
|
||||
Usage: "check health status",
|
||||
Category: "info",
|
||||
Before: func(c *cli.Context) error {
|
||||
err := parser.ParseConfig(cfg)
|
||||
if err != nil {
|
||||
fmt.Printf("%v", err)
|
||||
}
|
||||
return err
|
||||
},
|
||||
Action: func(c *cli.Context) error {
|
||||
logger := logging.Configure(cfg.Service.Name, cfg.Log)
|
||||
|
||||
resp, err := http.Get(
|
||||
fmt.Sprintf(
|
||||
"http://%s/healthz",
|
||||
cfg.Debug.Addr,
|
||||
),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
logger.Fatal().
|
||||
Err(err).
|
||||
Msg("Failed to request health check")
|
||||
}
|
||||
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
logger.Fatal().
|
||||
Int("code", resp.StatusCode).
|
||||
Msg("Health seems to be in bad state")
|
||||
}
|
||||
|
||||
logger.Debug().
|
||||
Int("code", resp.StatusCode).
|
||||
Msg("Health got a good state")
|
||||
|
||||
return nil
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package command
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config"
|
||||
"github.com/owncloud/ocis/ocis-pkg/clihelper"
|
||||
ociscfg "github.com/owncloud/ocis/ocis-pkg/config"
|
||||
"github.com/thejerf/suture/v4"
|
||||
"github.com/urfave/cli/v2"
|
||||
)
|
||||
|
||||
// GetCommands provides all commands for this service
|
||||
func GetCommands(cfg *config.Config) cli.Commands {
|
||||
return []*cli.Command{
|
||||
// start this service
|
||||
Server(cfg),
|
||||
|
||||
// interaction with this service
|
||||
|
||||
// infos about this service
|
||||
Health(cfg),
|
||||
Version(cfg),
|
||||
}
|
||||
}
|
||||
|
||||
// Execute is the entry point for the ocis-storage-metadata command.
|
||||
func Execute(cfg *config.Config) error {
|
||||
app := clihelper.DefaultApp(&cli.App{
|
||||
Name: "ocis-storage-metadata",
|
||||
Usage: "Provide apps for oCIS",
|
||||
Commands: GetCommands(cfg),
|
||||
})
|
||||
|
||||
cli.HelpFlag = &cli.BoolFlag{
|
||||
Name: "help,h",
|
||||
Usage: "Show the help",
|
||||
}
|
||||
|
||||
return app.Run(os.Args)
|
||||
}
|
||||
|
||||
// SutureService allows for the storage-metadata command to be embedded and supervised by a suture supervisor tree.
|
||||
type SutureService struct {
|
||||
cfg *config.Config
|
||||
}
|
||||
|
||||
// NewSutureService creates a new storage-metadata.SutureService
|
||||
func NewSutureService(cfg *ociscfg.Config) suture.Service {
|
||||
cfg.StorageMetadata.Commons = cfg.Commons
|
||||
return SutureService{
|
||||
cfg: cfg.StorageMetadata,
|
||||
}
|
||||
}
|
||||
|
||||
func (s SutureService) Serve(ctx context.Context) error {
|
||||
s.cfg.Context = ctx
|
||||
if err := Execute(s.cfg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
package command
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path"
|
||||
|
||||
"github.com/cs3org/reva/v2/cmd/revad/runtime"
|
||||
"github.com/gofrs/uuid"
|
||||
"github.com/oklog/run"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config/parser"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/logging"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/revaconfig"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/server/debug"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/tracing"
|
||||
"github.com/owncloud/ocis/ocis-pkg/service/external"
|
||||
"github.com/owncloud/ocis/ocis-pkg/sync"
|
||||
"github.com/owncloud/ocis/ocis-pkg/version"
|
||||
"github.com/urfave/cli/v2"
|
||||
)
|
||||
|
||||
// Server is the entry point for the server command.
|
||||
func Server(cfg *config.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "server",
|
||||
Usage: fmt.Sprintf("start %s extension without runtime (unsupervised mode)", cfg.Service.Name),
|
||||
Category: "server",
|
||||
Before: func(c *cli.Context) error {
|
||||
err := parser.ParseConfig(cfg)
|
||||
if err != nil {
|
||||
fmt.Printf("%v", err)
|
||||
}
|
||||
return err
|
||||
},
|
||||
Action: func(c *cli.Context) error {
|
||||
logger := logging.Configure(cfg.Service.Name, cfg.Log)
|
||||
err := tracing.Configure(cfg, logger)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
gr := run.Group{}
|
||||
ctx, cancel := defineContext(cfg)
|
||||
|
||||
defer cancel()
|
||||
|
||||
pidFile := path.Join(os.TempDir(), "revad-"+cfg.Service.Name+"-"+uuid.Must(uuid.NewV4()).String()+".pid")
|
||||
|
||||
rcfg := revaconfig.StorageMetadataFromStruct(cfg)
|
||||
|
||||
gr.Add(func() error {
|
||||
runtime.RunWithOptions(rcfg, pidFile, runtime.WithLogger(&logger.Logger))
|
||||
return nil
|
||||
}, func(_ error) {
|
||||
logger.Info().
|
||||
Str("server", cfg.Service.Name).
|
||||
Msg("Shutting down server")
|
||||
|
||||
cancel()
|
||||
})
|
||||
|
||||
debugServer, err := debug.Server(
|
||||
debug.Logger(logger),
|
||||
debug.Context(ctx),
|
||||
debug.Config(cfg),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
logger.Info().Err(err).Str("server", "debug").Msg("Failed to initialize server")
|
||||
return err
|
||||
}
|
||||
|
||||
gr.Add(debugServer.ListenAndServe, func(_ error) {
|
||||
cancel()
|
||||
})
|
||||
|
||||
if !cfg.Supervised {
|
||||
sync.Trap(&gr, cancel)
|
||||
}
|
||||
|
||||
if err := external.RegisterGRPCEndpoint(
|
||||
ctx,
|
||||
cfg.GRPC.Namespace+"."+cfg.Service.Name,
|
||||
uuid.Must(uuid.NewV4()).String(),
|
||||
cfg.GRPC.Addr,
|
||||
version.String,
|
||||
logger,
|
||||
); err != nil {
|
||||
logger.Fatal().Err(err).Msg("failed to register the grpc endpoint")
|
||||
}
|
||||
|
||||
if err := external.RegisterHTTPEndpoint(
|
||||
ctx,
|
||||
cfg.HTTP.Namespace+"."+cfg.Service.Name,
|
||||
uuid.Must(uuid.NewV4()).String(),
|
||||
cfg.HTTP.Addr,
|
||||
version.String,
|
||||
logger,
|
||||
); err != nil {
|
||||
logger.Fatal().Err(err).Msg("failed to register the http endpoint")
|
||||
}
|
||||
|
||||
return gr.Run()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// defineContext sets the context for the extension. If there is a context configured it will create a new child from it,
|
||||
// if not, it will create a root context that can be cancelled.
|
||||
func defineContext(cfg *config.Config) (context.Context, context.CancelFunc) {
|
||||
return func() (context.Context, context.CancelFunc) {
|
||||
if cfg.Context == nil {
|
||||
return context.WithCancel(context.Background())
|
||||
}
|
||||
return context.WithCancel(cfg.Context)
|
||||
}()
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
package command
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"github.com/owncloud/ocis/ocis-pkg/registry"
|
||||
"github.com/owncloud/ocis/ocis-pkg/version"
|
||||
|
||||
tw "github.com/olekukonko/tablewriter"
|
||||
"github.com/owncloud/ocis/extensions/storage-metadata/pkg/config"
|
||||
"github.com/urfave/cli/v2"
|
||||
)
|
||||
|
||||
// Version prints the service versions of all running instances.
|
||||
func Version(cfg *config.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "version",
|
||||
Usage: "print the version of this binary and the running extension instances",
|
||||
Category: "info",
|
||||
Action: func(c *cli.Context) error {
|
||||
fmt.Println("Version: " + version.String)
|
||||
fmt.Printf("Compiled: %s\n", version.Compiled())
|
||||
fmt.Println("")
|
||||
|
||||
reg := registry.GetRegistry()
|
||||
services, err := reg.GetService(cfg.GRPC.Namespace + "." + cfg.Service.Name)
|
||||
if err != nil {
|
||||
fmt.Println(fmt.Errorf("could not get %s services from the registry: %v", cfg.Service.Name, err))
|
||||
return err
|
||||
}
|
||||
|
||||
if len(services) == 0 {
|
||||
fmt.Println("No running " + cfg.Service.Name + " service found.")
|
||||
return nil
|
||||
}
|
||||
|
||||
table := tw.NewWriter(os.Stdout)
|
||||
table.SetHeader([]string{"Version", "Address", "Id"})
|
||||
table.SetAutoFormatHeaders(false)
|
||||
for _, s := range services {
|
||||
for _, n := range s.Nodes {
|
||||
table.Append([]string{s.Version, n.Address, n.Id})
|
||||
}
|
||||
}
|
||||
table.Render()
|
||||
return nil
|
||||
},
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user