From 709c47af3e3ded0f0e48f949bbb938fd22d7b904 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adam=20=C5=A0trauch?= Date: Wed, 15 Sep 2021 16:58:09 +0200 Subject: [PATCH] Callback script Callback is run when there some discovery packet changes. It can be an update, adding a new one or deleting the old one. This can be used to perform dynamic configuration of services that don't support lobby's API. --- README.md | 49 ++++++++++++++------------- daemon/config.go | 3 ++ daemon/main.go | 26 ++++++++++++-- daemon/update.go | 86 +++++++++++++++++++++++++++++++++++++++++++++++ templater/main.go | 38 +++++++++++++++++++++ 5 files changed, 177 insertions(+), 25 deletions(-) create mode 100644 templater/main.go diff --git a/README.md b/README.md index 122802a..9c74b2f 100644 --- a/README.md +++ b/README.md @@ -89,29 +89,32 @@ To test if local instance is running call this: There are other config directives you can use to fine-tune lobbyd to exactly what you need. -| Environment variable | Type | Default | Required | Note | -| ----------------------- | ------ | ----------------- | ----------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | -| TOKEN | string | | no | Authentication token for API, if empty auth is disabled | -| HOST | string | 127.0.0.1 | no | IP address used for the REST server to listen | -| PORT | int | 1313 | no | Port related to the address above | -| DISABLE_API | bool | false | no | If true API interface won't start | -| DRIVER | string | NATS | yes | Selects which driver is used to exchange the discovery packets. | -| NATS_URL | string | | yes (NATS driver) | NATS URL used to connect to the NATS server | -| NATS_DISCOVERY_CHANNEL | string | lobby.discovery | no | Channel where the keep-alive packets are sent | -| REDIS_HOST | string | 127.0.0.1" | no | Redis host | -| REDIS_PORT | uint16 | 6379 | no | Redis port | -| REDIS_DB | string | 0 | no | Redis DB | -| REDIS_CHANNEL | string | lobby:discovery | no | Redis channel | -| REDIS_PASSWORD | string | | no | Redis password | -| LABELS | string | | no | List of labels, labels should be separated by comma | -| LABELS_PATH | string | /etc/lobby/labels | no | Path where filesystem based labels are located, one label per line, filename is not important for lobby | -| RUNTIME_LABELS_FILENAME | string | _runtime | no | Filename for file created in LabelsPath where runtime labels will be added | -| HOSTNAME | string | | no | Override local machine's hostname | -| CLEAN_EVERY | int | 15 | no | How often to clean the list of discovered servers to get rid of the not alive ones [secs] | -| KEEP_ALIVE | int | 5 | no | how often to send the keep-alive discovery message with all available information [secs] | -| TTL | int | 30 | no | After how many secs is discovery record considered as invalid | -| NODE_EXPORTER_PORT | int | 9100 | no | Default port where node_exporter listens on all registered servers, this is used when the special prometheus labels doesn't contain port | -| REGISTER | bool | true | no | If true (default) then local instance is registered with other instance (discovery packet is sent regularly), if false the daemon runs only as a client | +| Environment variable | Type | Default | Required | Note | +| ------------------------ | ------ | ----------------- | ----------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | +| TOKEN | string | | no | Authentication token for API, if empty auth is disabled | +| HOST | string | 127.0.0.1 | no | IP address used for the REST server to listen | +| PORT | int | 1313 | no | Port related to the address above | +| DISABLE_API | bool | false | no | If true API interface won't start | +| DRIVER | string | NATS | yes | Selects which driver is used to exchange the discovery packets. | +| NATS_URL | string | | yes (NATS driver) | NATS URL used to connect to the NATS server | +| NATS_DISCOVERY_CHANNEL | string | lobby.discovery | no | Channel where the keep-alive packets are sent | +| REDIS_HOST | string | 127.0.0.1" | no | Redis host | +| REDIS_PORT | uint16 | 6379 | no | Redis port | +| REDIS_DB | string | 0 | no | Redis DB | +| REDIS_CHANNEL | string | lobby:discovery | no | Redis channel | +| REDIS_PASSWORD | string | | no | Redis password | +| LABELS | string | | no | List of labels, labels should be separated by comma | +| LABELS_PATH | string | /etc/lobby/labels | no | Path where filesystem based labels are located, one label per line, filename is not important for lobby | +| RUNTIME_LABELS_FILENAME | string | _runtime | no | Filename for file created in LabelsPath where runtime labels will be added | +| HOSTNAME | string | | no | Override local machine's hostname | +| CLEAN_EVERY | int | 15 | no | How often to clean the list of discovered servers to get rid of the not alive ones [secs] | +| KEEP_ALIVE | int | 5 | no | how often to send the keep-alive discovery message with all available information [secs] | +| TTL | int | 30 | no | After how many secs is discovery record considered as invalid | +| NODE_EXPORTER_PORT | int | 9100 | no | Default port where node_exporter listens on all registered servers, this is used when the special prometheus labels doesn't contain port | +| REGISTER | bool | true | no | If true (default) then local instance is registered with other instance (discovery packet is sent regularly), if false the daemon runs only as a client | +| CALLBACK | string | | no | Path to a script that runs when the the discovery packet records are changed. Not running for first | +| CALLBACK_COOLDOWN | int | 15 | no | Cooldown prevents the call back script to run sooner than configured amount of seconds after last run is finished. | +| CALLBACK_FIRST_RUN_DELAY | int | 30 | no | Wait for this amount of seconds before callback is run for first time after fresh start of the daemon | ### Service discovery for Prometheus diff --git a/daemon/config.go b/daemon/config.go index c0c098a..534ddaf 100644 --- a/daemon/config.go +++ b/daemon/config.go @@ -30,6 +30,9 @@ type Config struct { TTL uint `envconfig:"TTL" required:"false" default:"30"` // After how many secs is discovery record considered as invalid NodeExporterPort uint `envconfig:"NODE_EXPORTER_PORT" required:"false" default:"9100"` // Default port where node_exporter listens on all registered servers Register bool `envconfig:"REGISTER" required:"false" default:"true"` // If true (default) then local instance is registered with other instance (discovery packet is sent regularly) + Callback string `envconfig:"CALLBACK" required:"false" default:""` // path to a script that runs when the is a change in the labels database + CallbackCooldown uint `envconfig:"CALLBACK_COOLDOWN" required:"false" default:"15"` // cooldown that prevents to run the config change script too many times in row + CallbackFirstRunDelay uint `envconfig:"CALLBACK_FIRST_RUN_DELAY" required:"false" default:"30"` // Wait for this amount of seconds before callback is run for first time after fresh start of the daemon } // GetConfig return configuration created based on environment variables diff --git a/daemon/main.go b/daemon/main.go index 07f1bcb..4908edb 100644 --- a/daemon/main.go +++ b/daemon/main.go @@ -20,6 +20,7 @@ import ( var discoveryStorage server.Discoveries = server.Discoveries{} var driver common.Driver var localHost server.LocalHost +var lastLocalHostname string var config Config @@ -103,6 +104,7 @@ func sendDiscoveryPacketTask(trigger chan bool) { <-trigger if !shuttingDown { + // Get info about local machine and send to the exchange point discovery, err := localHost.GetIdentification() if err != nil { log.Printf("sending discovery identification error: %v\n", err) @@ -112,6 +114,19 @@ func sendDiscoveryPacketTask(trigger chan bool) { if err != nil { log.Println(err.Error()) } + + // If local hostname changes, we will deregister it + if discovery.Hostname != lastLocalHostname && lastLocalHostname != "" { + log.Println("Hostname change detected, deregistering the old one") + err = driver.SendGoodbyePacket(server.Discovery{ + Hostname: lastLocalHostname, + }) + if err != nil { + log.Println(err) + } + } + + lastLocalHostname = discovery.Hostname } else { break } @@ -139,7 +154,7 @@ func main() { // Setup callback function to register and unregister discovery packets from other servers driver.RegisterSubscribeFunction(func(d server.Discovery) { - // Check if the local version and the new version are somehow changed + // Check if the local version and the new version are somehowe changed localVersion := discoveryStorage.Get(d.Hostname) exists := discoveryStorage.Exist(d.Hostname) changed := server.Compare(localVersion, d) @@ -147,7 +162,6 @@ func main() { discoveryStorage.Add(d) if changed { - // Print this only if the server is already registered if exists { log.Printf("%s has been updated", d.Hostname) @@ -179,6 +193,14 @@ func main() { go printDiscoveryLogs() go cleanDiscoveryPool() + go discoveryChangeLoop() + go changeCatcherLoop() + + // When the daemon boots up we trigger discovery change event so the config can be setup via callback script if there is any + err = discoveryChange(server.Discovery{}) + if err != nil { + log.Printf("discovery changed error: %v", err) + } // If config.Register is false this instance won't be registered with other nodes if config.Register { diff --git a/daemon/update.go b/daemon/update.go index a014689..a98f25f 100644 --- a/daemon/update.go +++ b/daemon/update.go @@ -1,14 +1,100 @@ package main import ( + "encoding/json" + "log" + "os/exec" + "strings" + "time" + "github.com/by-cx/lobby/server" ) // These functions are called when something has changed in the storage +var changeDetectedChannel chan bool = make(chan bool) +var changeDetected bool + +// changeCatcherLoop waits for a change signal and switches variable that says if the callback should run or not +func changeCatcherLoop() { + for { + <-changeDetectedChannel + changeDetected = true + } +} + +// discoveryChangeLoop is used to process dynamic configuration changes. +// When there is a change detected a shell script or given process is triggered +// which does some operations with the new data. Usually it generates the +// configuration. +// This function has internal loop and won't allow to run the command more +// often than it's the configured amount of time. That prevents +func discoveryChangeLoop() { + // This other loop tics in strict intervals and prevents the callback script to run more often than it's configured + var cmd *exec.Cmd + + // Delay first run of the callback script a little so everything can set up + log.Printf("Delaying start of discovery change loop (%d seconds)\n", config.CallbackFirstRunDelay) + time.Sleep(time.Duration(config.CallbackFirstRunDelay) * time.Second) + log.Println("Starting discovery change loop") + + for { + if changeDetected { + // We switch this at the beginning so we can detect new changes while the callback script is running + changeDetected = false + + log.Println("Running callback function") + + // TODO: this is not the best way + callbackCommandSlice := strings.Split(config.Callback, " ") + + if len(callbackCommandSlice) == 1 { + cmd = exec.Command(callbackCommandSlice[0]) + } else if len(callbackCommandSlice) > 1 { + cmd = exec.Command(callbackCommandSlice[0], callbackCommandSlice[1:]...) + } else { + log.Println("wrong number of parts of the callback command") + time.Sleep(time.Duration(config.CallbackCooldown) * time.Second) + continue + } + + stdin, err := cmd.StdinPipe() + if err != nil { + log.Println("stdin writing error: ", err.Error()) + continue + } + + discoveriesJSON, err := json.Marshal(discoveryStorage.GetAll()) + if err != nil { + log.Println("stdin writing error: ", err.Error()) + continue + } + + _, err = stdin.Write([]byte(discoveriesJSON)) + if err != nil { + log.Println("stdin writing error: ", err.Error()) + continue + } + err = stdin.Close() + if err != nil { + log.Println("stdin writing error: ", err.Error()) + continue + } + + stdout, err := cmd.CombinedOutput() + if err != nil { + log.Println("Callback error: ", err.Error()) + } + log.Println("Callback output: ", string(stdout)) + } + time.Sleep(time.Duration(config.CallbackCooldown) * time.Second) + } +} + // discoveryChange is called when daemon detects that a newly arrived discovery // packet is somehow different than the localone. This can be used to trigger // some action in the local machine. func discoveryChange(discovery server.Discovery) error { + changeDetectedChannel <- true return nil } diff --git a/templater/main.go b/templater/main.go new file mode 100644 index 0000000..3024367 --- /dev/null +++ b/templater/main.go @@ -0,0 +1,38 @@ +package main + +import ( + "flag" + "fmt" + "os" +) + +// Templater is used to generate config files from predefined +// templates based on content of gathered discovery packets. +// It can for example configure Nginx's backend or database +// replication. +// +// It reads templates from /var/lib/lobby/templates (default) +// which are YAML files cotaining the template itself and command(s) +// that needs to be run when the template changes. + +const defaultTemplatesPath = "/var/lib/lobby/templates" + +var templatesPath *string + +func init() { + templatesPath = flag.String("templates-path", defaultTemplatesPath, "path of where templates are stored") + + flag.Parse() + + err := os.MkdirAll(*templatesPath, 0750) + if err != nil { + fmt.Println(err) + flag.Usage() + os.Exit(1) + } + +} + +func main() { + fmt.Println(*templatesPath) +}