docker: remove flags, enable env vars

Signed-off-by: Sander Pick <[email protected]>

.gitignore

@@ -13,11 +13,11 @@
 
 # Dependency directories (remove the comment below to include it)
 vendor/
-service/pb/github.com
 
 # Examples repo
 .threads*/
 .e2e_counter_writeraddr
+docker-compose.yml
 
 # JS PB files
 *.js

Dockerfile

@@ -75,4 +75,4 @@ VOLUME $THREADS_PATH
 
 ENTRYPOINT ["/sbin/tini", "--", "threadsd"]
 
-CMD ["--repo=/data/threads", "--serviceApiAddr=/ip4/0.0.0.0/tcp/5006", "--serviceApiProxyAddr=/ip4/0.0.0.0/tcp/5007", "--apiAddr=/ip4/0.0.0.0/tcp/6006", "--apiProxyAddr=/ip4/0.0.0.0/tcp/6007"]
+CMD ["--repo=/data/threads"]

go.mod

@@ -47,6 +47,7 @@ require (
  github.com/multiformats/go-multihash v0.0.10
  github.com/multiformats/go-varint v0.0.2 // indirect
  github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f // indirect
+ github.com/namsral/flag v1.7.4-pre // indirect
  github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
  github.com/polydawn/refmt v0.0.0-20190807091052-3d65705ee9f1 // indirect
  github.com/rs/cors v1.7.0 // indirect

go.sum

@@ -509,6 +509,8 @@ github.com/multiformats/go-varint v0.0.2 h1:6sUvyh2YHpJCb8RZ6eYzj6iJQ4+chWYmyIHx
 github.com/multiformats/go-varint v0.0.2/go.mod h1:3Ls8CIEsrijN6+B7PbrXRPxHRPuXSrVKRY101jdMZYE=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f h1:KUppIJq7/+SVif2QVs3tOP0zanoHgBEVAwHxUSIzRqU=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/namsral/flag v1.7.4-pre h1:b2ScHhoCUkbsq0d2C15Mv+VU8bl8hAXV8arnWiOHNZs=
+github.com/namsral/flag v1.7.4-pre/go.mod h1:OXldTctbM6SWH1K899kPZcf65KxJiD7MsceFUpB5yDo=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=

jsonpatcher/jsonpatcher.go

@@ -120,10 +120,10 @@ func (jp *jsonPatcher) Reduce(
  return nil, errCantCreateExistingInstance
  }
  if err := txn.Put(key, je.Patch.JSONPatch); err != nil {
- return nil, fmt.Errorf("error when reducing create event: %v", err)
+ return nil, fmt.Errorf("error when reducing create event: %w", err)
  }
  if err := indexFunc(e.Model(), key, nil, je.Patch.JSONPatch, txn); err != nil {
- return nil, fmt.Errorf("error when indexing created data: %v", err)
+ return nil, fmt.Errorf("error when indexing created data: %w", err)
  }
  actions[i] = core.ReduceAction{Type: core.Create, Model: e.Model(), EntityID: e.EntityID()}
  log.Debug("\tcreate operation applied")
@@ -137,13 +137,13 @@ func (jp *jsonPatcher) Reduce(
  }
  patchedValue, err := jsonpatch.MergePatch(value, je.Patch.JSONPatch)
  if err != nil {
- return nil, fmt.Errorf("error when reducing save event: %v", err)
+ return nil, fmt.Errorf("error when reducing save event: %w", err)
  }
  if err = txn.Put(key, patchedValue); err != nil {
  return nil, err
  }
  if err := indexFunc(e.Model(), key, value, patchedValue, txn); err != nil {
- return nil, fmt.Errorf("error when indexing created data: %v", err)
+ return nil, fmt.Errorf("error when indexing created data: %w", err)
  }
  actions[i] = core.ReduceAction{Type: core.Save, Model: e.Model(), EntityID: e.EntityID()}
  log.Debug("\tsave operation applied")
@@ -156,7 +156,7 @@ func (jp *jsonPatcher) Reduce(
  return nil, err
  }
  if err := indexFunc(e.Model(), key, value, nil, txn); err != nil {
- return nil, fmt.Errorf("error when removing index: %v", err)
+ return nil, fmt.Errorf("error when removing index: %w", err)
  }
  actions[i] = core.ReduceAction{Type: core.Delete, Model: e.Model(), EntityID: e.EntityID()}
  log.Debug("\tdelete operation applied")

service/client.go

@@ -97,7 +97,7 @@ func (s *server) pushLog(ctx context.Context, id thread.ID, lg thread.LogInfo, p
  client := pb.NewServiceClient(conn)
  _, err = client.PushLog(cctx, lreq)
  if err != nil {
- log.Warnf("push log to %s failed: %s", pid.String(), err)
+ return fmt.Errorf("push log to %s failed: %s", pid.String(), err)
  }
  return err
 }

service/service.go

@@ -414,6 +414,8 @@ func (t *service) AddFollower(ctx context.Context, id thread.ID, paddr ma.Multia
  dialable, err := getDialable(paddr)
  if err == nil {
  t.host.Peerstore().AddAddr(pid, dialable, pstore.PermanentAddrTTL)
+ } else {
+ log.Warnf("peer %s address requires a DHT lookup", pid.String())
  }
 
  // Send all logs to the new follower

store/index.go

@@ -25,9 +25,9 @@ const (
 // ErrUniqueExists is the error thrown when data is being inserted for a unique constraint value that already exists
 var (
  indexPrefix = ds.NewKey("_index")
- ErrUniqueExists = errors.New("Unique constraint violation")
- ErrNotIndexable = errors.New("Value not indexable")
- ErrNoIndexFound = errors.New("No index found")
+ ErrUniqueExists = errors.New("unique constraint violation")
+ ErrNotIndexable = errors.New("value not indexable")
+ ErrNoIndexFound = errors.New("no index found")
 )
 
 // Indexer is the interface to implement to support Model property indexes
@@ -166,7 +166,7 @@ func (v *keyList) in(key ds.Key) bool {
  return bytes.Compare((*v)[i], b) >= 0
  })
 
- return (i < len(*v) && bytes.Equal((*v)[i], b))
+ return i < len(*v) && bytes.Equal((*v)[i], b)
 }
 
 type MarshaledResult struct {

store/store.go

@@ -69,21 +69,6 @@ type Store struct {
  stateChangedNotifee *stateChangedNotifee
 }
 
-func defaultIndexFunc(s *Store) func(model string, key ds.Key, oldData, newData []byte, txn ds.Txn) error {
- return func(model string, key ds.Key, oldData, newData []byte, txn ds.Txn) error {
- indexer := s.GetModel(model)
- if err := indexDelete(indexer, txn, key, oldData); err != nil {
- return err
- }
- if newData != nil {
- if err := indexAdd(indexer, txn, key, newData); err != nil {
- return err
- }
- }
- return nil
- }
-}
-
 // NewStore creates a new Store, which will *own* ds and dispatcher for internal use.
 // Saying it differently, ds and dispatcher shouldn't be used externally.
 func NewStore(ts service.Service, opts ...Option) (*Store, error) {
@@ -152,10 +137,10 @@ func (s *Store) reregisterSchemas() error {
 
  for res := range results.Next() {
  name := ds.RawKey(res.Key).Name()
- index, err := s.datastore.Get(dsStoreIndexes.ChildString(name))
  var indexes []*IndexConfig
+ index, err := s.datastore.Get(dsStoreIndexes.ChildString(name))
  if err == nil && index != nil {
- json.Unmarshal(index, &indexes)
+ _ = json.Unmarshal(index, &indexes)
  }
  if _, err := s.RegisterSchema(name, string(res.Value), indexes...); err != nil {
  return err
@@ -287,7 +272,7 @@ func (s *Store) RegisterSchema(name string, schema string, indexes ...*IndexConf
 
  for _, config := range indexes {
  // @todo: Should check to make sure this is a valid field path for this schema
- m.AddIndex(config.Path, config.Unique)
+ _ = m.AddIndex(config.Path, config.Unique)
  }
 
  indexBytes, err := json.Marshal(indexes)
@@ -425,3 +410,18 @@ func isValidModel(t interface{}) (valid bool) {
  }
  return v.Elem().FieldByName(idFieldName).IsValid()
 }
+
+func defaultIndexFunc(s *Store) func(model string, key ds.Key, oldData, newData []byte, txn ds.Txn) error {
+ return func(model string, key ds.Key, oldData, newData []byte, txn ds.Txn) error {
+ indexer := s.GetModel(model)
+ if err := indexDelete(indexer, txn, key, oldData); err != nil {
+ return err
+ }
+ if newData != nil {
+ if err := indexAdd(indexer, txn, key, newData); err != nil {
+ return err
+ }
+ }
+ return nil
+ }
+}

threadsd/main.go

@@ -2,11 +2,12 @@ package main
 
 import (
  "context"
- "flag"
  "fmt"
+ "os"
 
  logging "github.com/ipfs/go-log"
  ma "github.com/multiformats/go-multiaddr"
+ "github.com/namsral/flag"
  "github.com/textileio/go-threads/api"
  serviceapi "github.com/textileio/go-threads/service/api"
  "github.com/textileio/go-threads/store"
@@ -16,14 +17,18 @@ import (
 var log = logging.Logger("threadsd")
 
 func main() {
- repo := flag.String("repo", ".threads", "repo location")
- hostAddrStr := flag.String("hostAddr", "/ip4/0.0.0.0/tcp/4006", "Threads host bind address")
- serviceApiAddrStr := flag.String("serviceApiAddr", "/ip4/127.0.0.1/tcp/5006", "Threads service API bind address")
- serviceApiProxyAddrStr := flag.String("serviceApiProxyAddr", "/ip4/127.0.0.1/tcp/5007", "Threads service API gRPC proxy bind address")
- apiAddrStr := flag.String("apiAddr", "/ip4/127.0.0.1/tcp/6006", "API bind address")
- apiProxyAddrStr := flag.String("apiProxyAddr", "/ip4/127.0.0.1/tcp/6007", "API gRPC proxy bind address")
- debug := flag.Bool("debug", false, "Enable debug logging")
- flag.Parse()
+ fs := flag.NewFlagSetWithEnvPrefix(os.Args[0], "THRDS", 0)
+
+ repo := fs.String("repo", ".threads", "repo location")
+ hostAddrStr := fs.String("hostAddr", "/ip4/0.0.0.0/tcp/4006", "Threads host bind address")
+ serviceApiAddrStr := fs.String("serviceApiAddr", "/ip4/127.0.0.1/tcp/5006", "Threads service API bind address")
+ serviceApiProxyAddrStr := fs.String("serviceApiProxyAddr", "/ip4/127.0.0.1/tcp/5007", "Threads service API gRPC proxy bind address")
+ apiAddrStr := fs.String("apiAddr", "/ip4/127.0.0.1/tcp/6006", "API bind address")
+ apiProxyAddrStr := fs.String("apiProxyAddr", "/ip4/127.0.0.1/tcp/6007", "API gRPC proxy bind address")
+ debug := fs.Bool("debug", false, "Enable debug logging")
+ if err := fs.Parse(os.Args[1:]); err != nil {
+ log.Fatal(err)
+ }
 
  hostAddr, err := ma.NewMultiaddr(*hostAddrStr)
  if err != nil {
@@ -53,6 +58,14 @@ func main() {
  }
  }
 
+ log.Debugf("repo: %v", *repo)
+ log.Debugf("hostAddr: %v", *hostAddrStr)
+ log.Debugf("serviceApiAddr: %v", *serviceApiAddrStr)
+ log.Debugf("serviceApiProxyAddr: %v", *serviceApiProxyAddrStr)
+ log.Debugf("apiAddr: %v", *apiAddrStr)
+ log.Debugf("apiProxyAddr: %v", *apiProxyAddrStr)
+ log.Debugf("debug: %v", *debug)
+
  ts, err := store.DefaultService(
  *repo,
  store.WithServiceHostAddr(hostAddr),