forked from nyaruka/rp-indexer
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
36db967
commit 3a051b3
Showing
5 changed files
with
154 additions
and
44 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| package indexer | ||
|
|
||
| import "os" | ||
|
|
||
| type Config struct { | ||
| ElasticURL string `help:"the url for our elastic search instance"` | ||
| DB string `help:"the connection string for our database"` | ||
| Index string `help:"the alias for our contact index"` | ||
| Poll int `help:"the number of seconds to wait between checking for updated contacts"` | ||
| Rebuild bool `help:"whether to rebuild the index, swapping it when complete, then exiting (default false)"` | ||
| Cleanup bool `help:"whether to remove old indexes after a rebuild"` | ||
| LogLevel string `help:"the log level, one of error, warn, info, debug"` | ||
| SentryDSN string `help:"the sentry configuration to log errors to, if any"` | ||
|
|
||
| LibratoUsername string `help:"the username that will be used to authenticate to Librato"` | ||
| LibratoToken string `help:"the token that will be used to authenticate to Librato"` | ||
| InstanceName string `help:"the unique name of this instance used for analytics"` | ||
| } | ||
|
|
||
| func NewDefaultConfig() *Config { | ||
| hostname, _ := os.Hostname() | ||
|
|
||
| return &Config{ | ||
| ElasticURL: "http://localhost:9200", | ||
| DB: "postgres://localhost/temba?sslmode=disable", | ||
| Index: "contacts", | ||
| Poll: 5, | ||
| Rebuild: false, | ||
| Cleanup: false, | ||
| LogLevel: "info", | ||
| InstanceName: hostname, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package indexer | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/nyaruka/librato" | ||
| "github.com/nyaruka/rp-indexer/indexers" | ||
| "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| type Daemon struct { | ||
| cfg *Config | ||
| db *sql.DB | ||
| wg *sync.WaitGroup | ||
| quit chan bool | ||
| indexers []indexers.Indexer | ||
| } | ||
|
|
||
| // NewDaemon creates a new daemon to run the given indexers | ||
| func NewDaemon(cfg *Config, db *sql.DB, ixs []indexers.Indexer) *Daemon { | ||
| return &Daemon{ | ||
| cfg: cfg, | ||
| db: db, | ||
| wg: &sync.WaitGroup{}, | ||
| quit: make(chan bool), | ||
| indexers: ixs, | ||
| } | ||
| } | ||
|
|
||
| // Start starts this daemon | ||
| func (d *Daemon) Start() { | ||
| // if we have a librato token, configure it | ||
| if d.cfg.LibratoToken != "" { | ||
| librato.Configure(d.cfg.LibratoUsername, d.cfg.LibratoToken, d.cfg.InstanceName, time.Second, d.wg) | ||
| librato.Start() | ||
| } | ||
|
|
||
| for _, i := range d.indexers { | ||
| d.startIndexer(i, time.Second*5) | ||
| } | ||
| } | ||
|
|
||
| func (d *Daemon) startIndexer(indexer indexers.Indexer, interval time.Duration) { | ||
| d.wg.Add(1) // add ourselves to the wait group | ||
|
|
||
| go func() { | ||
| defer func() { | ||
| logrus.WithField("indexer", indexer.Name()).Info("indexer exiting") | ||
| d.wg.Done() | ||
| }() | ||
|
|
||
| for { | ||
| select { | ||
| case <-d.quit: | ||
| return | ||
| case <-time.After(interval): | ||
| _, err := indexer.Index(d.db, d.cfg.Rebuild, d.cfg.Cleanup) | ||
| if err != nil { | ||
| logrus.WithField("index", d.cfg.Index).WithError(err).Error("error during indexing") | ||
| } | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| // Stop stops this daemon | ||
| func (d *Daemon) Stop() { | ||
| logrus.Info("daemon stopping") | ||
| librato.Stop() | ||
|
|
||
| close(d.quit) | ||
| d.wg.Wait() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters