Documentation updates and clarifications for the New Storage implementation

cmd/ignite/cmd/daemon.go

@@ -2,13 +2,13 @@ package cmd
 
 import (
  "fmt"
- "github.com/weaveworks/ignite/pkg/errutils"
  "io"
  "os"
  "os/signal"
  "sync"
 
  "github.com/spf13/cobra"
+ "github.com/weaveworks/ignite/pkg/errutils"
  "github.com/weaveworks/ignite/pkg/providers"
 )
 

pkg/storage/sync/storage.go

@@ -9,12 +9,13 @@ import (
  "github.com/weaveworks/ignite/pkg/storage"
  "github.com/weaveworks/ignite/pkg/storage/watch"
  "github.com/weaveworks/ignite/pkg/storage/watch/update"
+ "github.com/weaveworks/ignite/pkg/util"
  "k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 type UpdateStream chan update.Update
 
-const updateBuffer = 4096 // How many updates to buffer
+const updateBuffer = 4096 // How many updates to buffer, 4096 should be enough for even a high update frequency
 
 // SyncStorage is a Storage implementation taking in multiple Storages and
 // keeping them in sync. Any write operation executed on the SyncStorage
@@ -27,7 +28,7 @@ type SyncStorage struct {
  storages []storage.Storage
  eventStream watch.AssociatedEventStream
  updateStream UpdateStream
- monitor *watch.Monitor
+ monitor *util.Monitor
 }
 
 var _ storage.Storage = &SyncStorage{}
@@ -47,7 +48,7 @@ func NewSyncStorage(rwStorage storage.Storage, wStorages ...storage.Storage) sto
  }
 
  if ss.eventStream != nil {
- ss.monitor = watch.RunMonitor(ss.monitorFunc)
+ ss.monitor = util.RunMonitor(ss.monitorFunc)
  }
 
  return ss
@@ -141,7 +142,8 @@ func (ss *SyncStorage) monitorFunc() {
  case update.EventModify, update.EventCreate:
  // First load the Object using the Storage given in the update,
  // then set it using the client constructed above
- if obj, err := client.NewClient(upd.Storage).Dynamic(upd.APIType.GetKind()).Get(upd.APIType.GetUID()); err != nil {
+ updClient := client.NewClient(upd.Storage).Dynamic(upd.APIType.GetKind())
+ if obj, err := updClient.Get(upd.APIType.GetUID()); err != nil {
  log.Errorf("Failed to get Object with UID %q: %v", upd.APIType.GetUID(), err)
  continue
  } else if err = c.Dynamic(upd.APIType.GetKind()).Set(obj); err != nil {
@@ -156,8 +158,15 @@ func (ss *SyncStorage) monitorFunc() {
  }
  }
 
- // Send the update to the listeners
- ss.updateStream <- upd.Update
+ // Send the update to the listeners unless the channel is full,
+ // in which case issue a warning. The channel can hold as many
+ // updates as updateBuffer specifies.
+ select {
+ case ss.updateStream <- upd.Update:
+ log.Debugf("SyncStorage: sent update: %v", upd.Update)
+ default:
+ log.Warn("SyncStorage: failed to send update, channel full")
+ }
  } else {
  return
  }

pkg/storage/watch/storage.go

@@ -11,13 +11,16 @@ import (
  "github.com/weaveworks/ignite/pkg/storage"
  "github.com/weaveworks/ignite/pkg/storage/manifest/raw"
  "github.com/weaveworks/ignite/pkg/storage/watch/update"
+ "github.com/weaveworks/ignite/pkg/util"
  "k8s.io/apimachinery/pkg/runtime"
  "k8s.io/apimachinery/pkg/runtime/schema"
  "sigs.k8s.io/yaml"
 )
 
-// Storage is an interface for persisting and retrieving API objects to/from a backend
-// One Storage instance handles all different Kinds of Objects
+// WatchStorage is an extended Storage implementation, which provides a watcher
+// for watching changes in the directory managed by the embedded Storage's RawStorage.
+// If the RawStorage is a MappedRawStorage instance, it's mappings will automatically
+// be updated by the WatchStorage. Update events are sent to the given event stream.
 type WatchStorage interface {
  // WatchStorage extends the Storage interface
  storage.Storage
@@ -42,7 +45,7 @@ func NewGenericWatchStorage(storage storage.Storage) (WatchStorage, error) {
  }
 
  if mapped, ok := s.RawStorage().(raw.MappedRawStorage); ok {
- s.monitor = RunMonitor(func() {
+ s.monitor = util.RunMonitor(func() {
  s.monitorFunc(mapped, files) // Offload the file registration to the goroutine
  })
  }
@@ -55,7 +58,7 @@ type GenericWatchStorage struct {
  storage.Storage
  watcher *watcher
  events *AssociatedEventStream
- monitor *Monitor
+ monitor *util.Monitor
 }
 
 var _ WatchStorage = &GenericWatchStorage{}

pkg/storage/watch/update/associated.go

@@ -4,6 +4,9 @@ import (
  "github.com/weaveworks/ignite/pkg/storage"
 )
 
+// AssociatedUpdate bundles together an Update and a Storage
+// implementation. This is used by SyncStorage to query the
+// correct Storage for the updated Object.
 type AssociatedUpdate struct {
  Update
  Storage storage.Storage

pkg/storage/watch/update/event.go

@@ -1,6 +1,6 @@
 package update
 
-// Event is an enum describing a change in a file's state.
+// Event is an enum describing a change in a file's/Object's state.
 // Unknown state changes can be signaled with a zero value.
 type Event uint8
 
@@ -22,10 +22,3 @@ func (e Event) String() string {
 
  return "NONE"
 }
-
-// FileUpdate is used by watchers to
-// signal the state change of a file.
-type FileUpdate struct {
- Event Event
- Path string
-}

pkg/storage/watch/update/file.go

@@ -0,0 +1,8 @@
+package update
+
+// FileUpdate is used by watchers to
+// signal the state change of a file.
+type FileUpdate struct {
+ Event Event
+ Path string
+}

pkg/storage/watch/update/update.go

@@ -4,6 +4,8 @@ import (
  meta "github.com/weaveworks/ignite/pkg/apis/meta/v1alpha1"
 )
 
+// Update bundles an Event with an
+// APIType for Storage retrieval.
 type Update struct {
  Event Event
  APIType meta.Object

pkg/storage/watch/watch.go

@@ -8,6 +8,7 @@ import (
  log "github.com/sirupsen/logrus"
  "github.com/weaveworks/ignite/pkg/storage"
  "github.com/weaveworks/ignite/pkg/storage/watch/update"
+ "github.com/weaveworks/ignite/pkg/util"
 )
 
 const eventBuffer = 4096 // How many events and updates we can buffer before watching is interrupted
@@ -24,13 +25,17 @@ type eventStream chan notify.EventInfo
 type UpdateStream chan *update.FileUpdate
 type watches []string
 
+// watcher recursively monitors changes in files in the given directory
+// and sends out events based on their state changes. Only files conforming
+// to validSuffix are monitored. The watcher can be suspended for a single
+// event at a time to eliminate updates by WatchStorage causing a loop.
 type watcher struct {
  dir string
  events eventStream
  updates UpdateStream
  watches watches
  suspendEvent update.Event
- monitor *Monitor
+ monitor *util.Monitor
 }
 
 func (w *watcher) addWatch(path string) (err error) {
@@ -73,7 +78,7 @@ func newWatcher(dir string) (w *watcher, files []string, err error) {
  if err = w.start(&files); err != nil {
  notify.Stop(w.events)
  } else {
- w.monitor = RunMonitor(w.monitorFunc)
+ w.monitor = util.RunMonitor(w.monitorFunc)
  }
 
  return
@@ -173,6 +178,8 @@ func (w *watcher) close() {
  w.monitor.Wait()
 }
 
+// This enables a one-time suspend of the given event,
+// the watcher will skip the given event once
 func (w *watcher) suspend(updateEvent update.Event) {
  w.suspendEvent = updateEvent
 }
@@ -185,6 +192,8 @@ func convertEvent(event notify.Event) update.Event {
  return 0
 }
 
+// validSuffix is used to filter out all unsupported
+// files based on the extensions in storage.Formats
 func validSuffix(path string) bool {
  for suffix := range storage.Formats {
  if filepath.Ext(path) == suffix {

pkg/storage/watch/monitor.go → pkg/util/monitor.go

@@ -1,7 +1,11 @@
-package watch
+package util
 
 import "sync"
 
+// Monitor is a convenience wrapper around
+// starting a goroutine with a wait group,
+// which can be used to wait for the
+// goroutine to stop.
 type Monitor struct {
  wg *sync.WaitGroup
 }