mirror/opencloud
1
0
Fork 0
mirror of https://github.com/opencloud-eu/opencloud.git synced 2026-01-06 04:09:40 -06:00
Code Issues Packages Projects Releases Wiki Activity

add ctl command to resume postprocessing

Browse Source 
Signed-off-by: jkoberg <jkoberg@owncloud.com>
This commit is contained in:
jkoberg
2023-06-12 15:13:57 +02:00
parent 76141a492f
commit aac793a400
4 changed files with 91 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
services/postprocessing/pkg/command/postprocessing.go Normal file
View File

@@ -0,0 +1,58 @@
package command
import (
"fmt"
"time"
"github.com/cs3org/reva/v2/pkg/events"
"github.com/cs3org/reva/v2/pkg/utils"
"github.com/owncloud/ocis/v2/ocis-pkg/config/configlog"
"github.com/owncloud/ocis/v2/services/postprocessing/pkg/config"
"github.com/owncloud/ocis/v2/services/postprocessing/pkg/config/parser"
"github.com/urfave/cli/v2"
)
// RestartPostprocessing cli command to restart postprocessing
func RestartPostprocessing(cfg *config.Config) *cli.Command {
return &cli.Command{
Name: "restart",
Usage: "restart postprocessing for an uploadID",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "upload-id",
Aliases: []string{"u"},
Required: true,
Usage: "the uploadid to restart",
},
},
Before: func(c *cli.Context) error {
return configlog.ReturnFatal(parser.ParseConfig(cfg))
},
Action: func(c *cli.Context) error {
stream, err := getEventBus(cfg.Postprocessing.Events)
if err != nil {
return err
}
ev := events.ResumePostprocessing{
UploadID: c.String("upload-id"),
Timestamp: utils.TSNow(),
}
if err := events.Publish(stream, ev); err != nil {
fmt.Println(err)
return err
}
// go-micro nats implementation uses async publishing,
// therefore we need to manually wait.
//
// FIXME: upstream pr
//
// https://github.com/go-micro/plugins/blob/3e77393890683be4bacfb613bc5751867d584692/v4/events/natsjs/nats.go#L115
time.Sleep(5 * time.Second)
return nil
},
}
}

1
services/postprocessing/pkg/command/root.go
View File

@@ -15,6 +15,7 @@ func GetCommands(cfg *config.Config) cli.Commands {
Server(cfg),
// interaction with this service
RestartPostprocessing(cfg),
// infos about this service
Health(cfg),

59
services/postprocessing/pkg/command/server.go
View File

@@ -7,6 +7,7 @@ import (
"fmt"
"os"
"github.com/cs3org/reva/v2/pkg/events"
"github.com/cs3org/reva/v2/pkg/events/stream"
"github.com/cs3org/reva/v2/pkg/store"
"github.com/go-micro/plugins/v4/events/natsjs"
@@ -42,9 +43,6 @@ func Server(cfg *config.Config) *cli.Command {
gr = run.Group{}
logger = logging.Configure(cfg.Service.Name, cfg.Log)
evtsCfg = cfg.Postprocessing.Events
tlsConf *tls.Config
ctx, cancel = func() (context.Context, context.CancelFunc) {
if cfg.Context == nil {
return context.WithCancel(context.Background())
@@ -55,31 +53,7 @@ func Server(cfg *config.Config) *cli.Command {
defer cancel()
{
if evtsCfg.EnableTLS {
var rootCAPool *x509.CertPool
if evtsCfg.TLSRootCACertificate != "" {
rootCrtFile, err := os.Open(evtsCfg.TLSRootCACertificate)
if err != nil {
return err
}
rootCAPool, err = ociscrypto.NewCertPoolFromPEM(rootCrtFile)
if err != nil {
return err
}
evtsCfg.TLSInsecure = false
}
tlsConf = &tls.Config{
RootCAs: rootCAPool,
}
}
bus, err := stream.Nats(
natsjs.TLSConfig(tlsConf),
natsjs.Address(evtsCfg.Endpoint),
natsjs.ClusterID(evtsCfg.Cluster),
)
bus, err := getEventBus(cfg.Postprocessing.Events)
if err != nil {
return err
}
@@ -136,3 +110,32 @@ func Server(cfg *config.Config) *cli.Command {
},
}
}
func getEventBus(evtsCfg config.Events) (events.Stream, error) {
var tlsConf *tls.Config
if evtsCfg.EnableTLS {
var rootCAPool *x509.CertPool
if evtsCfg.TLSRootCACertificate != "" {
rootCrtFile, err := os.Open(evtsCfg.TLSRootCACertificate)
if err != nil {
return nil, err
}
rootCAPool, err = ociscrypto.NewCertPoolFromPEM(rootCrtFile)
if err != nil {
return nil, err
}
evtsCfg.TLSInsecure = false
}
tlsConf = &tls.Config{
RootCAs: rootCAPool,
}
}
return stream.Nats(
natsjs.TLSConfig(tlsConf),
natsjs.Address(evtsCfg.Endpoint),
natsjs.ClusterID(evtsCfg.Cluster),
)
}