arcad/emissary
16
0
Fork 0
Code Issues 14 Pull Requests Packages Projects Releases 3 Wiki Activity

feat: basic authorization model

Browse Source
This commit is contained in:
wpetit
2023-03-13 10:44:58 +01:00
parent 55db21ad23
commit fa36d55163
28 changed files with 589 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
internal/agent/controller/app/controller.go
View File

@ -11,16 +11,21 @@ import (
"forge.cadoles.com/Cadoles/emissary/internal/spec/app"
"forge.cadoles.com/arcad/edge/pkg/bundle"
"forge.cadoles.com/arcad/edge/pkg/storage/sqlite"
"github.com/mitchellh/hashstructure/v2"
"github.com/pkg/errors"
"gitlab.com/wpetit/goweb/logger"
)
type serverEntry struct {
SpecHash uint64
Server *Server
}
type Controller struct {
currentSpecRevision int
client *http.Client
downloadDir string
dataDir string
servers map[string]*Server
client *http.Client
downloadDir string
dataDir string
servers map[string]*serverEntry
}
// Name implements node.Controller.
@ -34,9 +39,9 @@ func (c *Controller) Reconcile(ctx context.Context, state *agent.State) error {
if err := state.GetSpec(app.NameApp, appSpec); err != nil {
if errors.Is(err, agent.ErrSpecNotFound) {
logger.Info(ctx, "could not find app spec, stopping all remaining apps")
logger.Info(ctx, "could not find app spec")
c.stopAllApps(ctx)
c.stopAllApps(ctx, appSpec)
return nil
}
@ -46,22 +51,20 @@ func (c *Controller) Reconcile(ctx context.Context, state *agent.State) error {
logger.Info(ctx, "retrieved spec", logger.F("spec", appSpec.SpecName()), logger.F("revision", appSpec.SpecRevision()))
if c.currentSpecRevision == appSpec.SpecRevision() {
logger.Info(ctx, "spec revision did not change, doing nothing")
return nil
}
c.updateApps(ctx, appSpec)
return nil
}
func (c *Controller) stopAllApps(ctx context.Context) {
for appID, server := range c.servers {
func (c *Controller) stopAllApps(ctx context.Context, spec *app.Spec) {
if len(c.servers) > 0 {
logger.Info(ctx, "stopping all apps")
}
for appID, entry := range c.servers {
logger.Info(ctx, "stopping app", logger.F("appID", appID))
if err := server.Stop(); err != nil {
if err := entry.Server.Stop(); err != nil {
logger.Error(
ctx, "error while stopping app",
logger.F("appID", appID),
@ -74,17 +77,15 @@ func (c *Controller) stopAllApps(ctx context.Context) {
}
func (c *Controller) updateApps(ctx context.Context, spec *app.Spec) {
hadError := false
// Stop and remove obsolete apps
for appID, server := range c.servers {
for appID, entry := range c.servers {
if _, exists := spec.Apps[appID]; exists {
continue
}
logger.Info(ctx, "stopping app", logger.F("appID", appID))
if err := server.Stop(); err != nil {
if err := entry.Server.Stop(); err != nil {
logger.Error(
ctx, "error while stopping app",
logger.F("gatewayID", appID),
@ -92,8 +93,6 @@ func (c *Controller) updateApps(ctx context.Context, spec *app.Spec) {
)
delete(c.servers, appID)
hadError = true
}
}
@ -103,20 +102,17 @@ func (c *Controller) updateApps(ctx context.Context, spec *app.Spec) {
if err := c.updateApp(ctx, appID, appSpec); err != nil {
logger.Error(appCtx, "could not update app", logger.E(errors.WithStack(err)))
hadError = true
continue
}
}
if !hadError {
c.currentSpecRevision = spec.SpecRevision()
logger.Info(ctx, "updating current spec revision", logger.F("revision", c.currentSpecRevision))
}
}
func (c *Controller) updateApp(ctx context.Context, appID string, appSpec app.AppEntry) error {
func (c *Controller) updateApp(ctx context.Context, appID string, appSpec app.AppEntry) (err error) {
newAppSpecHash, err := hashstructure.Hash(appSpec, hashstructure.FormatV2, nil)
if err != nil {
return errors.WithStack(err)
}
bundle, sha256sum, err := c.ensureAppBundle(ctx, appID, appSpec)
if err != nil {
return errors.Wrap(err, "could not download app bundle")
@ -127,7 +123,9 @@ func (c *Controller) updateApp(ctx context.Context, appID string, appSpec app.Ap
return errors.Wrap(err, "could not retrieve app data dir")
}
server, exists := c.servers[appID]
var entry *serverEntry
entry, exists := c.servers[appID]
if !exists {
logger.Info(ctx, "app currently not running")
} else if sha256sum != appSpec.SHA256Sum {
@ -137,35 +135,54 @@ func (c *Controller) updateApp(ctx context.Context, appID string, appSpec app.Ap
logger.F("specHash", appSpec.SHA256Sum),
)
if err := server.Stop(); err != nil {
if err := entry.Server.Stop(); err != nil {
return errors.Wrap(err, "could not stop app")
}
server = nil
entry = nil
}
if server == nil {
if entry == nil {
dbFile := filepath.Join(dataDir, appID+".sqlite")
db, err := sqlite.Open(dbFile)
if err != nil {
return errors.Wrapf(err, "could not opend database file '%s'", dbFile)
}
server = NewServer(bundle, db)
c.servers[appID] = server
entry = &serverEntry{
Server: NewServer(bundle, db),
SpecHash: 0,
}
c.servers[appID] = entry
}
logger.Info(
ctx, "starting app",
logger.F("address", appSpec.Address),
)
specChanged := newAppSpecHash != entry.SpecHash
if err := server.Start(ctx, appSpec.Address); err != nil {
if entry.Server.Running() && !specChanged {
return nil
}
if specChanged && entry.SpecHash != 0 {
logger.Info(
ctx, "restarting app",
logger.F("address", appSpec.Address),
)
} else {
logger.Info(
ctx, "starting app",
logger.F("address", appSpec.Address),
)
}
if err := entry.Server.Start(ctx, appSpec.Address); err != nil {
delete(c.servers, appID)
return errors.Wrap(err, "could not start app")
}
entry.SpecHash = newAppSpecHash
return nil
}
@ -275,11 +292,10 @@ func NewController(funcs ...OptionFunc) *Controller {
}
return &Controller{
client: opts.Client,
downloadDir: opts.DownloadDir,
dataDir: opts.DataDir,
currentSpecRevision: -1,
servers: make(map[string]*Server),
client: opts.Client,
downloadDir: opts.DownloadDir,
dataDir: opts.DataDir,
servers: make(map[string]*serverEntry),
}
}

60
internal/agent/controller/app/server.go
View File

@ -4,30 +4,35 @@ import (
"context"
"database/sql"
"net/http"
"sync"
"forge.cadoles.com/arcad/edge/pkg/app"
"forge.cadoles.com/arcad/edge/pkg/bus"
"forge.cadoles.com/arcad/edge/pkg/bus/memory"
edgeHTTP "forge.cadoles.com/arcad/edge/pkg/http"
"forge.cadoles.com/arcad/edge/pkg/module"
"forge.cadoles.com/arcad/edge/pkg/module/auth"
"forge.cadoles.com/arcad/edge/pkg/module/cast"
"forge.cadoles.com/arcad/edge/pkg/module/net"
"forge.cadoles.com/arcad/edge/pkg/storage"
"forge.cadoles.com/arcad/edge/pkg/storage/sqlite"
"gitlab.com/wpetit/goweb/logger"
"forge.cadoles.com/arcad/edge/pkg/bundle"
"github.com/dop251/goja"
"github.com/go-chi/chi/middleware"
"github.com/go-chi/chi/v5"
"github.com/pkg/errors"
)
type Server struct {
bundle bundle.Bundle
db *sql.DB
server *http.Server
bundle bundle.Bundle
db *sql.DB
server *http.Server
serverMutex sync.RWMutex
}
func (s *Server) Start(ctx context.Context, addr string) error {
func (s *Server) Start(ctx context.Context, addr string) (err error) {
if s.server != nil {
if err := s.Stop(); err != nil {
return errors.WithStack(err)
@ -58,6 +63,18 @@ func (s *Server) Start(ctx context.Context, addr string) error {
}
go func() {
defer func() {
if recovered := recover(); recovered != nil {
if err, ok := recovered.(error); ok {
logger.Error(ctx, err.Error(), logger.E(errors.WithStack(err)))
return
}
panic(recovered)
}
}()
defer func() {
if err := s.Stop(); err != nil {
panic(errors.WithStack(err))
@ -69,18 +86,29 @@ func (s *Server) Start(ctx context.Context, addr string) error {
}
}()
s.serverMutex.Lock()
s.server = server
s.serverMutex.Unlock()
return nil
}
func (s *Server) Running() bool {
s.serverMutex.RLock()
defer s.serverMutex.RUnlock()
return s.server != nil
}
func (s *Server) Stop() error {
if s.server == nil {
return nil
}
defer func() {
s.serverMutex.Lock()
s.server = nil
s.serverMutex.Unlock()
}()
if err := s.server.Close(); err != nil {
@ -100,20 +128,18 @@ func (s *Server) getAppModules(bus bus.Bus, ds storage.DocumentStore, bs storage
module.RPCModuleFactory(bus),
module.StoreModuleFactory(ds),
module.BlobModuleFactory(bus, bs),
// module.Extends(
// auth.ModuleFactory(
// auth.WithJWT(dummyKeyFunc),
// ),
// func(o *goja.Object) {
// if err := o.Set("CLAIM_ROLE", "role"); err != nil {
// panic(errors.New("could not set 'CLAIM_ROLE' property"))
// }
module.Extends(
auth.ModuleFactory(),
func(o *goja.Object) {
if err := o.Set("CLAIM_ROLE", "role"); err != nil {
panic(errors.New("could not set 'CLAIM_ROLE' property"))
}
// if err := o.Set("CLAIM_PREFERRED_USERNAME", "preferred_username"); err != nil {
// panic(errors.New("could not set 'CLAIM_PREFERRED_USERNAME' property"))
// }
// },
// ),
if err := o.Set("CLAIM_PREFERRED_USERNAME", "preferred_username"); err != nil {
panic(errors.New("could not set 'CLAIM_PREFERRED_USERNAME' property"))
}
},
),
}
}