Skip to content

Commit

Permalink
arch-211-add-logging-to-silo (#53)
Browse files Browse the repository at this point in the history 
* Added logging to migrator

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Sync now passes logger to migrator

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Updated logger module

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Added logging to device

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Added logging to nbd expose

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Added logging to cmd/serve

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Fix nbd config, and added logging output for tests

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Couple of logs in device.go

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

* Added logging to waitingCache

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>

---------

Signed-off-by: Jimmy Moore <jamesmoore@loopholelabs.io>
  • Loading branch information
@jimmyaxod
jimmyaxod authored Nov 19, 2024
1 parent a54a0e2 commit 8cafaa2
Show file tree
Hide file tree
Showing 20 changed files with 583 additions and 99 deletions.
2 changes: 1 addition & 1 deletion cmd/connect.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -356,7 +356,7 @@ func handleIncomingDevice(ctx context.Context, pro protocol.Protocol, dev uint32

// Called to setup an exposed storage device
func dstDeviceSetup(prov storage.Provider) (storage.ExposedStorage, error) {
p := expose.NewExposedStorageNBDNL(prov, 1, 0, prov.Size(), 4096, true)
p := expose.NewExposedStorageNBDNL(prov, expose.DefaultConfig)
var err error

err = p.Init()
Expand Down
30 changes: 21 additions & 9 deletions cmd/serve.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ import (
"time"

"github.com/fatih/color"
"github.com/loopholelabs/logging"
"github.com/loopholelabs/logging/types"
"github.com/loopholelabs/silo/pkg/storage"
"github.com/loopholelabs/silo/pkg/storage/blocks"
"github.com/loopholelabs/silo/pkg/storage/config"
Expand Down Expand Up @@ -48,13 +50,16 @@ var srcStorage []*storageInfo
var serveProgressBar *mpb.Progress
var serveBars []*mpb.Bar

var serveDebug bool

func init() {
rootCmd.AddCommand(cmdServe)
cmdServe.Flags().StringVarP(&serveAddr, "addr", "a", ":5170", "Address to serve from")
cmdServe.Flags().StringVarP(&serveConf, "conf", "c", "silo.conf", "Configuration file")
cmdServe.Flags().BoolVarP(&serveProgress, "progress", "p", false, "Show progress")
cmdServe.Flags().BoolVarP(&serveContinuous, "continuous", "C", false, "Continuous sync")
cmdServe.Flags().BoolVarP(&serveAnyOrder, "order", "o", false, "Any order (faster)")
cmdServe.Flags().BoolVarP(&serveDebug, "debug", "d", false, "Debug logging (trace)")
}

type storageInfo struct {
Expand All @@ -69,6 +74,12 @@ type storageInfo struct {
}

func runServe(_ *cobra.Command, _ []string) {
var log types.RootLogger
if serveDebug {
log = logging.New(logging.Zerolog, "silo.serve", os.Stderr)
log.SetLevel(types.TraceLevel)
}

if serveProgress {
serveProgressBar = mpb.New(
mpb.WithOutput(color.Output),
Expand All @@ -85,7 +96,7 @@ func runServe(_ *cobra.Command, _ []string) {
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
go func() {
<-c
shutdownEverything()
shutdownEverything(log)
os.Exit(1)
}()

Expand All @@ -96,7 +107,7 @@ func runServe(_ *cobra.Command, _ []string) {

for i, s := range siloConf.Device {
fmt.Printf("Setup storage %d [%s] size %s - %d\n", i, s.Name, s.Size, s.ByteSize())
sinfo, err := setupStorageDevice(s)
sinfo, err := setupStorageDevice(s, log)
if err != nil {
panic(fmt.Sprintf("Could not setup storage. %v", err))
}
Expand All @@ -108,7 +119,7 @@ func runServe(_ *cobra.Command, _ []string) {

l, err := net.Listen("tcp", serveAddr)
if err != nil {
shutdownEverything()
shutdownEverything(log)
panic("Listener issue...")
}

Expand All @@ -133,7 +144,7 @@ func runServe(_ *cobra.Command, _ []string) {
for i, s := range srcStorage {
wg.Add(1)
go func(index int, src *storageInfo) {
err := migrateDevice(uint32(index), src.name, pro, src)
err := migrateDevice(log, uint32(index), src.name, pro, src)
if err != nil {
fmt.Printf("There was an issue migrating the storage %d %v\n", index, err)
}
Expand All @@ -149,10 +160,10 @@ func runServe(_ *cobra.Command, _ []string) {

con.Close()
}
shutdownEverything()
shutdownEverything(log)
}

func shutdownEverything() {
func shutdownEverything(_ types.RootLogger) {
// first unlock everything
fmt.Printf("Unlocking devices...\n")
for _, i := range srcStorage {
Expand All @@ -169,8 +180,8 @@ func shutdownEverything() {
}
}

func setupStorageDevice(conf *config.DeviceSchema) (*storageInfo, error) {
source, ex, err := device.NewDevice(conf)
func setupStorageDevice(conf *config.DeviceSchema, log types.RootLogger) (*storageInfo, error) {
source, ex, err := device.NewDeviceWithLogging(conf, log)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -223,7 +234,7 @@ func setupStorageDevice(conf *config.DeviceSchema) (*storageInfo, error) {
}

// Migrate a device
func migrateDevice(devID uint32, name string,
func migrateDevice(log types.RootLogger, devID uint32, name string,
pro protocol.Protocol,
sinfo *storageInfo) error {
size := sinfo.lockable.Size()
Expand Down Expand Up @@ -293,6 +304,7 @@ func migrateDevice(devID uint32, name string,
}()

conf := migrator.NewConfig().WithBlockSize(sinfo.blockSize)
conf.Logger = log
conf.LockerHandler = func() {
_ = dest.SendEvent(&packets.Event{Type: packets.EventPreLock})
sinfo.lockable.Lock()
Expand Down
5 changes: 1 addition & 4 deletions cmd/sync.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -260,10 +260,7 @@ func syncMigrateS3(_ uint32, name string,
fmt.Printf("[%s] Error for block %d error %v\n", name, b.Block, err)
}

// Show logging for S3 writes
logDest := modules.NewLogger(destMetrics, "S3")

mig, err := migrator.NewMigrator(sinfo.tracker, logDest, sinfo.orderer, conf)
mig, err := migrator.NewMigrator(sinfo.tracker, destMetrics, sinfo.orderer, conf)

if err != nil {
return err
Expand Down
3 changes: 2 additions & 1 deletion go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ require (
github.com/spf13/cobra v1.8.1
github.com/stretchr/testify v1.9.0
github.com/vbauerster/mpb/v8 v8.8.3
golang.org/x/sys v0.25.0
)

require (
Expand Down Expand Up @@ -64,6 +65,7 @@ require (
github.com/rivo/uniseg v0.4.7 // indirect
github.com/rogpeppe/go-internal v1.10.0 // indirect
github.com/rs/xid v1.6.0 // indirect
github.com/rs/zerolog v1.33.0 // indirect
github.com/sirupsen/logrus v1.9.3 // indirect
github.com/spf13/pflag v1.0.5 // indirect
github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
Expand All @@ -74,7 +76,6 @@ require (
golang.org/x/mod v0.17.0 // indirect
golang.org/x/net v0.28.0 // indirect
golang.org/x/sync v0.8.0 // indirect
golang.org/x/sys v0.25.0 // indirect
golang.org/x/text v0.17.0 // indirect
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
Expand Down
7 changes: 7 additions & 0 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK3
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
github.com/containerd/continuity v0.4.3 h1:6HVkalIp+2u1ZLH1J/pYX2oBVXlJZvh1X1A7bEZ9Su8=
github.com/containerd/continuity v0.4.3/go.mod h1:F6PTNCKepoxEaXLQp3wDAjygEnImnZ/7o4JzpodfroQ=
github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY=
Expand Down Expand Up @@ -51,6 +52,7 @@ github.com/go-test/deep v1.0.3 h1:ZrJSEWsXzPOxaZnFteGEfooLba+ju3FYIbOrS+rQd68=
github.com/go-test/deep v1.0.3/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
github.com/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
Expand Down Expand Up @@ -86,6 +88,7 @@ github.com/loopholelabs/logging v0.3.1/go.mod h1:uRDUydiqPqKbZkb0WoQ3dfyAcJ2iOMh
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
Expand Down Expand Up @@ -127,8 +130,11 @@ github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUc
github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=
github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
github.com/rs/zerolog v1.33.0 h1:1cU2KZkvPxNyfgEmhHAz/1A9Bz+llsdYzklWFzgp0r8=
github.com/rs/zerolog v1.33.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
Expand Down Expand Up @@ -184,6 +190,7 @@ golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
Expand Down
41 changes: 39 additions & 2 deletions pkg/storage/device/device.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (
"sync"
"time"

"github.com/loopholelabs/logging/types"
"github.com/loopholelabs/silo/pkg/storage"
"github.com/loopholelabs/silo/pkg/storage/blocks"
"github.com/loopholelabs/silo/pkg/storage/config"
Expand All @@ -37,9 +38,13 @@ type Device struct {
}

func NewDevices(ds []*config.DeviceSchema) (map[string]*Device, error) {
return NewDevicesWithLogging(ds, nil)
}

func NewDevicesWithLogging(ds []*config.DeviceSchema, log types.RootLogger) (map[string]*Device, error) {
devices := make(map[string]*Device)
for _, c := range ds {
dev, ex, err := NewDevice(c)
dev, ex, err := NewDeviceWithLogging(c, log)
if err != nil {
// Close/shutdown any we already setup, but we'll ignore any close errors here.
for _, cc := range devices {
Expand All @@ -59,6 +64,13 @@ func NewDevices(ds []*config.DeviceSchema) (map[string]*Device, error) {
}

func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorage, error) {
return NewDeviceWithLogging(ds, nil)
}

func NewDeviceWithLogging(ds *config.DeviceSchema, log types.RootLogger) (storage.Provider, storage.ExposedStorage, error) {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("creating new device")
}

var prov storage.Provider
var err error
Expand Down Expand Up @@ -175,6 +187,9 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag

// Optionally use a copy on write RO source...
if ds.ROSource != nil {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("setting up CopyOnWrite")
}

// Create the ROSource...
rodev, _, err := NewDevice(ds.ROSource)
Expand All @@ -201,6 +216,10 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag

// Make sure the cow data gets dumped on close...
cow.CloseFn = func() {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("Writing CopyOnWrite state")
}

blocks := cow.GetBlockExists()
// Write it out to file
data := make([]byte, 0)
Expand All @@ -216,6 +235,10 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag

// Optionally binlog this dev to a file
if ds.Binlog != "" {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("logging to binlog")
}

prov, err = modules.NewBinLog(prov, ds.Binlog)
if err != nil {
return nil, nil, err
Expand All @@ -227,7 +250,7 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag
var ex storage.ExposedStorage
if ds.Expose {

ex = expose.NewExposedStorageNBDNL(prov, 8, 0, prov.Size(), expose.NBDDefaultBlockSize, true)
ex = expose.NewExposedStorageNBDNL(prov, expose.DefaultConfig.WithLogger(log))

err := ex.Init()
if err != nil {
Expand All @@ -238,6 +261,9 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag

// Optionally sync the device to S3
if ds.Sync != nil {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("setting up S3 sync")
}

s3dest, err := sources.NewS3StorageCreate(ds.Sync.Secure,
ds.Sync.Endpoint,
Expand Down Expand Up @@ -280,6 +306,7 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag

// Start doing the sync...
syncer := migrator.NewSyncer(ctx, &migrator.SyncConfig{
Logger: log,
Name: ds.Name,
Integrity: false,
CancelWrites: true,
Expand All @@ -306,6 +333,9 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag
var wg sync.WaitGroup

startSync := func(_ storage.EventType, data storage.EventData) storage.EventReturnData {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("sync.start called")
}
if data != nil {
startConfig := data.(storage.SyncStartConfig)

Expand Down Expand Up @@ -356,6 +386,9 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag
}

stopSync := func(_ storage.EventType, _ storage.EventData) storage.EventReturnData {
if log != nil {
log.Debug().Str("name", ds.Name).Msg("sync.stop called")
}
syncLock.Lock()
if !syncRunning {
syncLock.Unlock()
Expand All @@ -381,6 +414,10 @@ func NewDevice(ds *config.DeviceSchema) (storage.Provider, storage.ExposedStorag
altSources = append(altSources, as)
}

if log != nil {
log.Debug().Str("name", ds.Name).Int("sources", len(altSources)).Msg("sync.stop returning altSources")
}

return altSources
}

Expand Down
Loading

0 comments on commit 8cafaa2

Please sign in to comment.