terraform/schema: Implement watcher for invalidating cached schema

Closes #16

go.mod

@@ -5,13 +5,15 @@ go 1.13
 require (
  github.com/apparentlymart/go-textseg v1.0.0
  github.com/creachadair/jrpc2 v0.6.1
+ github.com/fsnotify/fsnotify v1.4.9
  github.com/google/go-cmp v0.4.0
  github.com/hashicorp/go-version v1.2.0
  github.com/hashicorp/hcl/v2 v2.3.0
  github.com/hashicorp/terraform-json v0.4.0
  github.com/mitchellh/cli v1.0.0
  github.com/sourcegraph/go-lsp v0.0.0-20200117082640-b19bb38222e2
  github.com/zclconf/go-cty v1.2.1
+ golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 // indirect
  golang.org/x/text v0.3.2
 )
 

go.sum

@@ -18,6 +18,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
+github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 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/golang/protobuf v1.1.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -74,6 +76,10 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190502175342-a43fa875dd82 h1:vsphBvatvfbhlb4PO1BYSr9dzugGxJ/SQHoNufZJq1w=
 golang.org/x/sys v0.0.0-20190502175342-a43fa875dd82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9 h1:L2auWcuQIvxz9xSEqzESnV/QN/gNRXNApHi3fYwl2w0=
+golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 h1:uYVVQ9WP/Ds2ROhcaGPeIdVq0RIXVLwsHlnvJ+cT1So=
+golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=

internal/terraform/lang/provider_block_test.go

@@ -288,12 +288,7 @@ func TestProviderBlock_CompletionItemsAtPos(t *testing.T) {
  pf.InitializeCapabilities(*caps)
  }
 
- var sr schema.Reader
- if tc.ps != nil {
- sr = schema.MockStorage(tc.ps)
- } else {
- sr = schema.MockStorage(&tfjson.ProviderSchemas{})
- }
+ sr := schema.MockStorage(tc.ps)
  pf.schemaReader = sr
 
  p, err := pf.New(block)

internal/terraform/schema/schema_storage.go

@@ -4,6 +4,7 @@ import (
  "fmt"
  "io/ioutil"
  "log"
+ "sync"
  "time"
 
  "github.com/hashicorp/go-version"
@@ -16,33 +17,63 @@ type Reader interface {
 }
 
 type Writer interface {
- ObtainSchemasForDir(*exec.Executor, string) error
- ProviderConfigSchema(name string) (*tfjson.Schema, error)
+ ObtainSchemasForWorkspace(*exec.Executor, string) error
+ AddWorkspaceForWatching(string) error
+ StartWatching(*exec.Executor) error
 }
 
-type storage struct {
- ps *tfjson.ProviderSchemas
+type Storage struct {
+ ps *tfjson.ProviderSchemas
+ w watcher
+ watching bool
 
  logger *log.Logger
+
+ // mu ensures atomic reading and obtaining of schemas
+ // as the process of obtaining it may not be thread-safe
+ mu sync.RWMutex
+
+ // sync makes operations synchronous which makes testing easier
+ sync bool
 }
 
-func NewStorage() *storage {
- return &storage{
- logger: log.New(ioutil.Discard, "", 0),
+var defaultLogger = log.New(ioutil.Discard, "", 0)
+
+func NewStorage() *Storage {
+ return &Storage{
+ logger: defaultLogger,
  }
 }
 
-func (s *storage) SetLogger(logger *log.Logger) {
+func (s *Storage) SetLogger(logger *log.Logger) {
  s.logger = logger
 }
 
-func MockStorage(ps *tfjson.ProviderSchemas) *storage {
- s := NewStorage()
- s.ps = ps
- return s
+// ObtainSchemasForWorkspace will (by default) asynchronously obtain schema via tf
+// and store it for later consumption via Reader methods
+func (s *Storage) ObtainSchemasForWorkspace(tf *exec.Executor, dir string) error {
+ if s.sync {
+ return s.obtainSchemasForWorkspace(tf, dir)
+ }
+
+ // This routine is not cancellable in itself
+ // but the time-consuming part is done by exec.Executor
+ // which is cancellable via its own context
+ go func() {
+ err := s.obtainSchemasForWorkspace(tf, dir)
+ if err != nil {
+ s.logger.Println("error obtaining schemas:", err)
+ }
+ }()
+
+ return nil
 }
 
-func (c *storage) ObtainSchemasForDir(tf *exec.Executor, dir string) error {
+func (s *Storage) obtainSchemasForWorkspace(tf *exec.Executor, dir string) error {
+ s.logger.Printf("Obtaining lock before retrieving schema for %q ...", dir)
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
  // Checking the version here may be excessive
  // TODO: Find a way to centralize this
  tfVersions, err := version.NewConstraint(">= 0.12.0")
@@ -56,20 +87,24 @@ func (c *storage) ObtainSchemasForDir(tf *exec.Executor, dir string) error {
 
  tf.SetWorkdir(dir)
 
- c.logger.Printf("Obtaining schemas for %q ...", dir)
+ s.logger.Printf("Retrieving schemas for %q ...", dir)
  start := time.Now()
  ps, err := tf.ProviderSchemas()
  if err != nil {
- return fmt.Errorf("unable to get schemas: %s", err)
+ return fmt.Errorf("Unable to retrieve schemas: %s", err)
  }
- c.ps = ps
- c.logger.Printf("Schemas retrieved in %s", time.Since(start))
-
+ s.ps = ps
+ s.logger.Printf("Schemas retrieved in %s", time.Since(start))
  return nil
 }
 
-func (c *storage) ProviderConfigSchema(name string) (*tfjson.Schema, error) {
- schema, ok := c.ps.Schemas[name]
+func (s *Storage) ProviderConfigSchema(name string) (*tfjson.Schema, error) {
+ s.logger.Printf("Obtaining lock before reading %q provider schema", name)
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+
+ s.logger.Printf("Reading %q provider schema", name)
+ schema, ok := s.ps.Schemas[name]
  if !ok {
  return nil, &SchemaUnavailableErr{"provider", name}
  }
@@ -80,3 +115,66 @@ func (c *storage) ProviderConfigSchema(name string) (*tfjson.Schema, error) {
 
  return schema.ConfigSchema, nil
 }
+
+// watcher creates a new Watcher instance
+// if one doesn't exist yet or returns an existing one
+func (s *Storage) watcher() (watcher, error) {
+ if s.w != nil {
+ return s.w, nil
+ }
+
+ w, err := NewWatcher()
+ if err != nil {
+ return nil, err
+ }
+ w.SetLogger(s.logger)
+
+ s.w = w
+ return s.w, nil
+}
+
+// StartWatching starts to watch for plugin changes in dirs that were added
+// via AddWorkspaceForWatching until StopWatching() is called
+func (s *Storage) StartWatching(tf *exec.Executor) error {
+ if s.watching {
+ return fmt.Errorf("watching already in progress")
+ }
+ w, err := s.watcher()
+ if err != nil {
+ return err
+ }
+
+ go w.OnPluginChange(func(ww *watchedWorkspace) error {
+ s.obtainSchemasForWorkspace(tf, ww.dir)
+ return nil
+ })
+ s.watching = true
+
+ s.logger.Printf("Watching for plugin changes ...")
+
+ return nil
+}
+
+func (s *Storage) StopWatching() error {
+ if s.w == nil {
+ return nil
+ }
+ s.logger.Println("Stopping watcher ...")
+ err := s.w.Close()
+ if err == nil {
+ s.watching = false
+ }
+
+ return err
+}
+
+func (s *Storage) AddWorkspaceForWatching(dir string) error {
+ w, err := s.watcher()
+ if err != nil {
+ return err
+ }
+
+ s.logger.Printf("Adding workspace for watching: %q", dir)
+
+ return w.AddWorkspace(dir)
+}

internal/terraform/schema/storage_mock.go

@@ -0,0 +1,41 @@
+package schema
+
+import (
+ "log"
+
+ "github.com/fsnotify/fsnotify"
+ tfjson "github.com/hashicorp/terraform-json"
+)
+
+func MockStorage(ps *tfjson.ProviderSchemas) *Storage {
+ s := NewStorage()
+ if ps == nil {
+ ps = &tfjson.ProviderSchemas{}
+ }
+ s.ps = ps
+ s.sync = true
+ s.w = &MockWatcher{}
+ return s
+}
+
+type MockWatcher struct{}
+
+func (w *MockWatcher) AddWorkspace(string) error {
+ return nil
+}
+
+func (w *MockWatcher) Close() error {
+ return nil
+}
+
+func (w *MockWatcher) Events() chan fsnotify.Event {
+ return nil
+}
+
+func (w *MockWatcher) Errors() chan error {
+ return nil
+}
+
+func (w *MockWatcher) OnPluginChange(func(*watchedWorkspace) error) {}
+
+func (w *MockWatcher) SetLogger(*log.Logger) {}