README
¶
vivoupdater
Vivoupdater subscribes to a Kafka Topic containing recently loaded Vivo triples. It will post into both Vivo and Vivo Widgets to trigger selective re-indexing of relevant people and organizations.
Dependencies
Dependencies are managed with Go Modules
To install:
go get github.com/OIT-ADS-Web/vivoupdater
cd $GOPATH/src/github.com/OIT-ADS-Web/vivoupdater
go install github.com/OIT-ADS-Web/vivoupdater...
This will create $GOPATH/bin/vivo_indexer and $GOPATH/bin/fake_produce.
Configuration
The following environment variables are used to control behavior:
APP_ENVIRONMENT
The environment the code is running in (e.g. development|acceptance|production). Also determines where Vault looks for keys (see next)
VAULT_ENDPOINT
The endpoint api of your vault installation
VAULT_KEY
Using the app_role auth method - this is the key value
VAULT_ROLE_ID
The vault role id
VAULT_SECRET_ID
The vault secret id value
BATCH_SIZE
default = 200
BATCH_TIMEOUT
default = 10
VIVO_INDEXER_URL
Where the indexer should post vivo updates
For example:
http://localhost:9080/searchService/updateUrisInSearch
VIVO_EMAIL
The vivo instance admin email, for example:
VIVO_PASSWORD
The vivo instance admin password
WIDGETS_INDEXER_BASE_URL
Where the indexer should post widgets updates
this will be be appened to with either a /person or /organization specification
for example:
http://localhost:8080/widgets/updates
WIDGETS_USER
widgets instance user
WIDGETS_PASSWORD
widgets instance password
BOOTSTRAP_SERVERS
The Kafka list of servers (as comma separated list)
UPDATES_TOPIC
The Kafka topic that has the updates - just one at this point
METRICS_TOPIC
The kafka topic to send metrics
CLIENT_ID
A kafka client name
GROUP_NAME
A kafka group name
Debugging
I've found pprof helpful. You can send in the flag pprof like this:
vivo_indexer -pprof=true
and it will enable (see go pprof):
For example (with Graphviz installed):
go tool pprof -png http://localhost:8484/debug/pprof/heap > out.png
Documentation
¶
Index ¶
- Constants
- Variables
- func FetchSecrets(config *VaultConfig, paths map[string]string, obj interface{}) error
- func FetchToken(config *VaultConfig) error
- func FluxLine(measurement string, c interface{}, tags map[string]string) (bytes.Buffer, error)
- func GetCertsFromVault(env string, config *VaultConfig, kafka *KafkaSubscriber)
- func GetProducer() sarama.AsyncProducer
- func IndexBatch(idx BatchIndexer, b map[string]bool, logger *log.Logger)
- func NewTLSConfig(ks *KafkaSubscriber) (*tls.Config, error)
- func PostToVivo(idx VivoIndexer, batch map[string]bool) error
- func PostToWidgets(idx WidgetsIndexer, postUrl string, uris ...string) error
- func Produce(topic string, val string)
- func SecretsMap(appEnv string) map[string]string
- func SendMetrics(metrics IndexMetrics)
- func SetProducer(p sarama.AsyncProducer)
- func SetSubscriber(s *KafkaSubscriber)
- func SetupConsumer(ks *KafkaSubscriber) error
- func SetupProducer(ks *KafkaSubscriber) error
- func StartConsumer(ctx context.Context, ks KafkaSubscriber, handler ConsumerGroupHandler) error
- type BatchIndexer
- type CSV
- type ConsumerGroupHandler
- type IndexMetrics
- type Indexer
- type KafkaConfig
- type KafkaSubscriber
- type Secrets
- type Triple
- type UpdateMessage
- type UpdateSubscriber
- type UriBatcher
- type VaultConfig
- type VaultData
- type Version
- type VivoIndexer
- type WidgetsBatchIndexer
- type WidgetsIndexer
- type WidgetsUpdateMessage
Constants ¶
const OrgFilterRegex = `.*individual/org[0-9]{8}`
const PersonFilterRegex = `.*individual/per[0-9A-Za-z]{3,}`
Variables ¶
var AppEnvironment string
var BatchSize int
misc
var BatchTimeout int
var ClientCert string
var ClientId string
var ClientKey string
var GroupName string
var LogFile string
logging
var LogMaxAge int
var LogMaxBackups int
var LogMaxSize int
var MetricsTopic string
var NotificationFrom string
var NotificationSmtp string
TODO: get rid of notification values entirely?
var ServerCert string
var UpdatesTopic string
var Topics CSV
var VaultEndpoint string
vault
var VaultKey string
var VaultRoleId string
var VaultSecretId string
var VivoEmail string
var VivoIndexerUrl string
vivo
var VivoPassword string
var WidgetsIndexerBaseUrl string
widgets
var WidgetsPassword string
var WidgetsUser string
Functions ¶
func FetchSecrets ¶
func FetchSecrets(config *VaultConfig, paths map[string]string, obj interface{}) error
TODO: this is rather convoluted bdcause it matches the methodology used by scholars-data-project kotlin client which maps to java properties one at a time (rather than as group of key/value)
So the SecretMap can (in theory) map a local property to a completely different vault secret e.g. kafka.clientKey = apps/scholars/development/kafka/anotherNameEntirely
func FetchToken ¶
func FetchToken(config *VaultConfig) error
func GetCertsFromVault ¶
func GetCertsFromVault(env string, config *VaultConfig, kafka *KafkaSubscriber)
func GetProducer ¶
func GetProducer() sarama.AsyncProducer
func IndexBatch ¶
func IndexBatch(idx BatchIndexer, b map[string]bool, logger *log.Logger)
func NewTLSConfig ¶
func NewTLSConfig(ks *KafkaSubscriber) (*tls.Config, error)
func PostToVivo ¶
func PostToVivo(idx VivoIndexer, batch map[string]bool) error
func PostToWidgets ¶
func PostToWidgets(idx WidgetsIndexer, postUrl string, uris ...string) error
NOTE: idx has username, password (and base url)
func SecretsMap ¶
func SendMetrics ¶
func SendMetrics(metrics IndexMetrics)
func SetProducer ¶
func SetProducer(p sarama.AsyncProducer)
func SetSubscriber ¶
func SetSubscriber(s *KafkaSubscriber)
func SetupProducer ¶
func SetupProducer(ks *KafkaSubscriber) error
func StartConsumer ¶
func StartConsumer(ctx context.Context, ks KafkaSubscriber, handler ConsumerGroupHandler) error
Types ¶
type BatchIndexer ¶
type CSV ¶
type CSV []string
stole code from here: https://godoc.org/github.com/namsral/flag
var BootstrapFlag CSV
kafka
var NotificationEmail CSV
type ConsumerGroupHandler ¶
type ConsumerGroupHandler struct {
Context context.Context
Logger *log.Logger
Updates chan UpdateMessage
Quit chan bool
}
func (ConsumerGroupHandler) Cleanup ¶
func (c ConsumerGroupHandler) Cleanup(sess sarama.ConsumerGroupSession) error
func (ConsumerGroupHandler) ConsumeClaim ¶
func (c ConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
func (ConsumerGroupHandler) Setup ¶
func (c ConsumerGroupHandler) Setup(sess sarama.ConsumerGroupSession) error
type IndexMetrics ¶
type KafkaConfig ¶
type KafkaSubscriber ¶
type KafkaSubscriber struct {
Brokers []string
Topic string
ClientCert string
ClientKey string
ServerCert string
ClientID string
GroupName string
}
var Subscriber *KafkaSubscriber
func GetSubscriber ¶
func GetSubscriber() *KafkaSubscriber
func (KafkaSubscriber) Subscribe ¶
func (ks KafkaSubscriber) Subscribe(ctx context.Context, logger *log.Logger, updates chan UpdateMessage, quit chan bool) error
TODO: seems like too many parameters
type UpdateMessage ¶
type UpdateSubscriber ¶
type UriBatcher ¶
type VaultConfig ¶
type VivoIndexer ¶
func (VivoIndexer) Name ¶
func (wi VivoIndexer) Name() string
type WidgetsBatchIndexer ¶
type WidgetsBatchIndexer struct {
Indexer WidgetsIndexer
Suffix string
Regex *regexp.Regexp
Uris []string
}
func NewWidgetsBatchIndexer ¶
func NewWidgetsBatchIndexer(wi WidgetsIndexer, suffix string, regex *regexp.Regexp) *WidgetsBatchIndexer
func (*WidgetsBatchIndexer) Gather ¶
func (wbi *WidgetsBatchIndexer) Gather(u string)
This pulls out URLs the particular indexer might be interested in typically they would look like one of the following:
https://scholars.duke.edu/individual/per3336942 https://scholars.duke.edu/individual/org50344699
type WidgetsIndexer ¶
func (WidgetsIndexer) Name ¶
func (wi WidgetsIndexer) Name() string
type WidgetsUpdateMessage ¶
type WidgetsUpdateMessage struct {
Uris []string `json:"uris"`
}
Source Files
Directories