mleku/indra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more refactoring

Browse Source
This commit is contained in:
greg stone
2023-02-27 18:17:09 +00:00
parent 1703cf3a79
commit cd741dd07a
15 changed files with 252 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/indra/root.go
View File

@@ -92,7 +92,7 @@ func initConfig() {
home, err := os.UserHomeDir()
cobra.CheckErr(err)
cfgFile = home + "/.indra.toml"
cfgFile = home + "/.indra/config.toml"
}
viper.SetConfigFile(cfgFile)

2
cmd/indra/seed_rpc.go
View File

@@ -44,7 +44,7 @@ var (
func initUnlock(cmd *cobra.Command) {
cmd.PersistentFlags().StringVarP(&unlockTarget, unlockTargetFlag, "",
cmd.Flags().StringVarP(&unlockTarget, unlockTargetFlag, "",
"unix:///tmp/indra.sock",
"the url of the rpc server",
)

3
cmd/indra/seed_serve.go
View File

@@ -39,6 +39,9 @@ var seedServeCommand = &cobra.Command{
go seed.Run(ctx)
select {
case err := <-seed.WhenStartFailed():
log.E.Ln("startup error:", err)
return
case <-seed.IsShutdown():
log.I.Ln("shutdown complete")
}

0
pkg/rpc/cmd.go → pkg/rpc/configure.go
View File

4
pkg/rpc/flags.go
View File

@@ -34,12 +34,12 @@ func InitFlags(cmd *cobra.Command) {
viper.BindPFlag(tunEnableFlag, cmd.PersistentFlags().Lookup(tunEnableFlag))
cmd.PersistentFlags().StringVarP(&tunKeyRaw, tunKeyFlag, "",
cmd.Flags().StringVarP(&tunKeyRaw, tunKeyFlag, "",
"",
"the base58 encoded pre-shared key for accessing the rpc",
)
viper.BindPFlag(tunKeyFlag, cmd.PersistentFlags().Lookup(tunKeyFlag))
viper.BindPFlag(tunKeyFlag, cmd.Flags().Lookup(tunKeyFlag))
cmd.PersistentFlags().IntVarP(&tunnelPort, tunPortFlag, "",
tunnelPort,

34
pkg/rpc/server.go
View File

@@ -1,7 +1,6 @@
package rpc
import (
"context"
"google.golang.org/grpc"
)
@@ -10,15 +9,19 @@ func init() {
}
var (
server *grpc.Server
startupErrors = make(chan error, 128)
isReady = make(chan bool, 1)
server *grpc.Server
)
func RunWith(ctx context.Context, r func(srv *grpc.Server)) {
func RunWith(r func(srv *grpc.Server), opts ...ServerOption) {
log.I.Ln("initializing the rpc server")
serverOpts := serverOptions{}
for _, opt := range opts {
opt.apply(&serverOpts)
}
configureUnixSocket()
configureTunnel()
@@ -26,18 +29,10 @@ func RunWith(ctx context.Context, r func(srv *grpc.Server)) {
log.I.Ln("starting rpc server")
go Start(ctx)
go Start()
}
func CantStart() chan error {
return startupErrors
}
func IsReady() chan bool {
return isReady
}
func Start(ctx context.Context) {
func Start() {
var err error
@@ -49,15 +44,12 @@ func Start(ctx context.Context) {
startupErrors <- err
}
isReady <- true
log.I.Ln("rpc server is ready")
select {
case <-ctx.Done():
Shutdown(context.Background())
}
isReady <- true
}
func Shutdown(ctx context.Context) {
func Shutdown() {
log.I.Ln("shutting down rpc server")

29
pkg/rpc/server_options.go Normal file
View File

@@ -0,0 +1,29 @@
package rpc
type serverOptions struct {
disableTunnel bool
}
type ServerOption interface {
apply(*serverOptions)
}
type funcServerOption struct {
f func(*serverOptions)
}
func (fdo *funcServerOption) apply(do *serverOptions) {
fdo.f(do)
}
func newFuncServerOption(f func(*serverOptions)) *funcServerOption {
return &funcServerOption{
f: f,
}
}
func WithDisableTunnel() ServerOption {
return newFuncServerOption(func(o *serverOptions) {
o.disableTunnel = true
})
}

14
pkg/rpc/signals.go Normal file
View File

@@ -0,0 +1,14 @@
package rpc
var (
startupErrors = make(chan error, 128)
isReady = make(chan bool, 1)
)
func WhenStartFailed() chan error {
return startupErrors
}
func IsReady() chan bool {
return isReady
}

80
pkg/seed/proc.go
View File

@@ -1,80 +0,0 @@
package seed
import (
"context"
"git-indra.lan/indra-labs/indra/pkg/storage"
"sync"
)
var (
inUse sync.Mutex
)
var (
startupErrors = make(chan error, 32)
isReadyChan = make(chan bool, 1)
isShutdownChan = make(chan bool, 1)
)
func CantStart() chan error {
return startupErrors
}
func IsReady() chan bool {
return isReadyChan
}
func IsShutdown() chan bool {
return isShutdownChan
}
func Shutdown() {
log.I.Ln("shutting down seed")
err := storage.Shutdown()
check(err)
isShutdownChan <- true
}
func Run(ctx context.Context) {
if !inUse.TryLock() {
log.E.Ln("seed is in use")
return
}
log.I.Ln("running seed")
var err error
go storage.Run(ctx)
signals:
for {
select {
case err = <-CantStart():
log.E.Ln("startup error:", err)
return
case <-storage.IsLocked():
log.I.Ln("storage is locked, waiting for unlock")
//go rpc.RunWith(ctx, func(srv *grpc.Server) {
// storage.RegisterUnlockServiceServer(srv, storage.NewUnlockService())
//})
// Run RPC unlock
case <-storage.IsReady():
break signals
case <-ctx.Done():
Shutdown()
return
}
}
log.I.Ln("seed is ready")
isReadyChan <- true
}

85
pkg/seed/server.go Normal file
View File

@@ -0,0 +1,85 @@
package seed
import (
"context"
"git-indra.lan/indra-labs/indra/pkg/rpc"
"git-indra.lan/indra-labs/indra/pkg/storage"
"google.golang.org/grpc"
"sync"
)
var (
inUse sync.Mutex
)
func Run(ctx context.Context) {
if !inUse.TryLock() {
log.E.Ln("seed is in use")
return
}
log.I.Ln("running seed")
go storage.Run()
signals:
for {
select {
case <-storage.WhenIsLocked():
log.I.Ln("storage is locked")
// Run an unlock RPC server
go rpc.RunWith(
func(srv *grpc.Server) {
storage.RegisterUnlockServiceServer(srv, storage.NewUnlockService())
},
rpc.WithDisableTunnel(),
)
case err := <-rpc.WhenStartFailed():
startupErrors <- err
case <-rpc.IsReady():
log.I.Ln("waiting for unlock")
case <-storage.WhenIsUnlocked():
log.I.Ln("restarting rpc server")
// Shut down unlock RPC server to we can launch the main one
rpc.Shutdown()
//go rpc.RunWith(func(srv *grpc.Server) {
// chat.RegisterChatServiceServer(srv, &chat.Server{})
//})
case <-storage.WhenIsReady():
break signals
case <-ctx.Done():
Shutdown()
return
}
}
log.I.Ln("seed is ready")
isReadyChan <- true
select {
case <-ctx.Done():
Shutdown()
return
}
}
func Shutdown() {
log.I.Ln("shutting down seed")
rpc.Shutdown()
err := storage.Shutdown()
check(err)
isShutdownChan <- true
}

19
pkg/seed/signals.go Normal file
View File

@@ -0,0 +1,19 @@
package seed
var (
startupErrors = make(chan error, 32)
isReadyChan = make(chan bool, 1)
isShutdownChan = make(chan bool, 1)
)
func WhenStartFailed() chan error {
return startupErrors
}
func IsReady() chan bool {
return isReadyChan
}
func IsShutdown() chan bool {
return isShutdownChan
}

14
pkg/storage/cmd.go → pkg/storage/configure.go
View File

@@ -7,10 +7,10 @@ import (
)
var (
isNewKey bool
isNewDB bool
isLocked bool
key Key
newKeyGenerated bool
isNewDB bool
noKeyProvided bool
key Key
)
func configure() {
@@ -71,15 +71,13 @@ func configureKey() {
log.I.Ln("no keyfile found")
isLocked = true
noKeyProvided = true
return
}
log.I.Ln("no keyfile found, generating a new key")
isNewKey = true
if key, err = KeyGen(); err != nil {
startupErrors <- err
return
@@ -115,6 +113,8 @@ func configureKey() {
log.W.Ln("-------------------------------------------------------")
log.W.Ln("")
newKeyGenerated = true
viper.Set(storeKeyFlag, key.Encode())
}

42
pkg/storage/flags.go
View File

@@ -6,29 +6,29 @@ import (
)
var (
storeKeyFlag = "store-key"
storeKeyFileFlag = "store-keyfile"
storeKeyRPCFlag = "store-key-rpc"
storeKeyFlag = "store-key"
storeKeyFileFlag = "store-keyfile"
//storeKeyRPCFlag = "store-key-rpc"
storeFilePathFlag = "store-path"
storeAskPassFlag = "store-ask-pass"
//storeAskPassFlag = "store-ask-pass"
)
var (
storeEncryptionKey string
storeEncryptionKeyFile string
storeEncryptionKeyRPC bool
storeFilePath string
storeAskPass bool
//storeEncryptionKeyRPC bool
storeFilePath string
//storeAskPass bool
)
func InitFlags(cmd *cobra.Command) {
cmd.PersistentFlags().StringVarP(&storeEncryptionKey, storeKeyFlag, "",
cmd.Flags().StringVarP(&storeEncryptionKey, storeKeyFlag, "",
"",
"the key required to unlock storage",
)
viper.BindPFlag(storeKeyFlag, cmd.PersistentFlags().Lookup(storeKeyFlag))
viper.BindPFlag(storeKeyFlag, cmd.Flags().Lookup(storeKeyFlag))
cmd.PersistentFlags().StringVarP(&storeEncryptionKeyFile, storeKeyFileFlag, "",
"",
@@ -44,17 +44,17 @@ func InitFlags(cmd *cobra.Command) {
viper.BindPFlag(storeFilePathFlag, cmd.PersistentFlags().Lookup(storeFilePathFlag))
cmd.PersistentFlags().BoolVarP(&storeEncryptionKeyRPC, storeKeyRPCFlag, "",
false,
"looks for the encryption key via RPC",
)
//cmd.PersistentFlags().BoolVarP(&storeEncryptionKeyRPC, storeKeyRPCFlag, "",
// false,
// "looks for the encryption key via RPC",
//)
//
//viper.BindPFlag(storeKeyRPCFlag, cmd.PersistentFlags().Lookup(storeKeyRPCFlag))
viper.BindPFlag(storeKeyRPCFlag, cmd.PersistentFlags().Lookup(storeKeyRPCFlag))
cmd.PersistentFlags().BoolVarP(&storeAskPass, storeAskPassFlag, "",
false,
"prompts the user for a password to unlock storage",
)
viper.BindPFlag(storeAskPassFlag, cmd.PersistentFlags().Lookup(storeAskPassFlag))
//cmd.PersistentFlags().BoolVarP(&storeAskPass, storeAskPassFlag, "",
// false,
// "prompts the user for a password to unlock storage",
//)
//
//viper.BindPFlag(storeAskPassFlag, cmd.PersistentFlags().Lookup(storeAskPassFlag))
}

99
pkg/storage/service.go
View File

@@ -1,8 +1,6 @@
package storage
import (
"context"
"git-indra.lan/indra-labs/indra/pkg/interrupt"
"github.com/dgraph-io/badger/v3"
"github.com/spf13/viper"
"sync"
@@ -10,26 +8,29 @@ import (
var (
fileName string = "indra.db"
db *badger.DB
opts badger.Options
)
var (
db *badger.DB
opts badger.Options
startupErrors = make(chan error, 128)
isLockedChan = make(chan bool, 1)
isReadyChan = make(chan bool, 1)
running sync.Mutex
)
func CantStart() chan error {
return startupErrors
}
func Run() {
func IsLocked() chan bool {
return isLockedChan
}
if !running.TryLock() {
return
}
func IsReady() chan bool {
return isReadyChan
configure()
if !attempt_unlock() {
isLockedChan <- true
return
}
log.I.Ln("storage is ready")
isReadyChan <- true
}
func Shutdown() (err error) {
@@ -40,6 +41,10 @@ func Shutdown() (err error) {
return nil
}
if db.IsClosed() {
return nil
}
if err = db.Close(); check(err) {
return
}
@@ -56,66 +61,28 @@ func Txn(tx func(txn *badger.Txn) error, update bool) error {
return tx(txn)
}
var (
running sync.Mutex
)
func attempt_unlock() bool {
func open() {
if noKeyProvided {
return false
}
var err error
log.I.Ln("attempting to unlock database")
opts = badger.DefaultOptions(viper.GetString(storeFilePathFlag))
opts.Logger = nil
opts.IndexCacheSize = 128 << 20
opts.EncryptionKey = key.Bytes()
if db, err = badger.Open(opts); check(err) {
startupErrors <- err
return
return false
}
log.I.Ln("successfully opened database")
log.I.Ln("storage is ready")
log.I.Ln("successfully unlocked database")
isUnlockedChan <- true
isReadyChan <- true
}
func Run(ctx context.Context) {
if !running.TryLock() {
return
}
configure()
opts = badger.DefaultOptions(viper.GetString(storeFilePathFlag))
opts.IndexCacheSize = 128 << 20
opts.Logger = nil
if !isLocked {
log.I.Ln("attempting to open database with key")
open()
}
isLockedChan <- true
lockedCtx, cancel := context.WithCancel(context.Background())
interrupt.AddHandler(cancel)
for {
select {
case <-IsReady():
log.I.Ln("storage is ready")
//case <-unlock.IsSuccessful():
//
// log.I.Ln("storage successfully unlocked")
//
// isReadyChan <- true
case <-lockedCtx.Done():
Shutdown()
return
}
}
return true
}

24
pkg/storage/signals.go Normal file
View File

@@ -0,0 +1,24 @@
package storage
var (
startupErrors = make(chan error, 128)
isLockedChan = make(chan bool, 1)
isUnlockedChan = make(chan bool, 1)
isReadyChan = make(chan bool, 1)
)
func CantStart() chan error {
return startupErrors
}
func WhenIsLocked() chan bool {
return isLockedChan
}
func WhenIsUnlocked() chan bool {
return isUnlockedChan
}
func WhenIsReady() chan bool {
return isReadyChan
}