Skip to content
This repository was archived by the owner on Jan 23, 2025. It is now read-only.

Make plugin retries configurable through input parameters #40

Merged
tbroden84 merged 6 commits into from
Dec 12, 2023
Merged

Make plugin retries configurable through input parameters #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9627125
Clean up and make plugins sleep/timeout configurable
tbroden84 Nov 23, 2023
83e1623
Clean up and make community plugins sleep/timeout configurable
tbroden84 Nov 23, 2023
769f174
Remove PluginParams struct
tbroden84 Nov 24, 2023
1db3c50
Update comment
tbroden84 Nov 24, 2023
b3f8aec
Merge branch 'main' into plugins-retries
tbroden84 Dec 12, 2023
d520ddf
Update with switch/case for response status code
tbroden84 Dec 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
214 changes: 131 additions & 83 deletions api/plugins.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,49 +7,39 @@ import (
"time"
)

type PluginParams struct {
Name string `json:"plugin_name,omitempty"`
Enabled bool `json:"enabled,omitempty"`
}

func (api *API) waitUntilPluginChanged(instanceID int, pluginName string, enabled bool) (map[string]interface{}, error) {
log.Printf("[DEBUG] go-api::plugin::waitUntilPluginChanged instance id: %v, name: %v", instanceID, pluginName)
for {
time.Sleep(10 * time.Second)
response, err := api.ReadPlugin(instanceID, pluginName)
log.Printf("[DEBUG] go-api::plugin::waitUntilPluginChanged response: %v", response)
if err != nil {
return nil, err
}
if response["required"] != nil && response["required"] != false {
return response, nil
}
if response["enabled"] == enabled {
return response, nil
}
}
}

func (api *API) EnablePlugin(instanceID int, pluginName string) (map[string]interface{}, error) {
failed := make(map[string]interface{})
params := &PluginParams{Name: pluginName}
log.Printf("[DEBUG] go-api::plugin::enable instance id: %v, params: %v", instanceID, pluginName)
path := fmt.Sprintf("/api/instances/%d/plugins?async=true", instanceID)
// EnablePlugin: enable a plugin on an instance.
func (api *API) EnablePlugin(instanceID int, pluginName string, sleep, timeout int) (
map[string]interface{}, error) {

var (
failed map[string]interface{}
params = make(map[string]interface{})
path = fmt.Sprintf("/api/instances/%d/plugins?async=true", instanceID)
)

params["plugin_name"] = pluginName
log.Printf("[DEBUG] go-api::plugin::enable instance id: %v, params: %v", instanceID, params)
response, err := api.sling.New().Post(path).BodyJSON(params).Receive(nil, &failed)

if err != nil {
return nil, err
}
if response.StatusCode != 204 {
return nil, fmt.Errorf("EnablePlugin failed, status: %v, message: %s", response.StatusCode, failed)
}

return api.waitUntilPluginChanged(instanceID, pluginName, true)
switch response.StatusCode {
case 204:
return api.waitUntilPluginChanged(instanceID, pluginName, true, 1, sleep, timeout)
default:
return nil,
fmt.Errorf("enable plugin failed, status: %v, message: %s", response.StatusCode, failed)
}
}

func (api *API) ReadPlugin(instanceID int, pluginName string) (map[string]interface{}, error) {
// ReadPlugin: reads a specific plugin from an instance.
func (api *API) ReadPlugin(instanceID int, pluginName string, sleep, timeout int) (
map[string]interface{}, error) {

log.Printf("[DEBUG] go-api::plugin::read instance id: %v, name: %v", instanceID, pluginName)
data, err := api.ReadPlugins(instanceID)
data, err := api.ListPlugins(instanceID, sleep, timeout)
if err != nil {
return nil, err
}
Expand All @@ -64,86 +54,144 @@ func (api *API) ReadPlugin(instanceID int, pluginName string) (map[string]interf
return nil, nil
}

func (api *API) ReadPlugins(instanceID int) ([]map[string]interface{}, error) {
// Initiale values, 5 attempts and 20 second sleep
return api.readPluginsWithRetry(instanceID, 5, 20)
// ListPlugins: list plugins from an instance.
func (api *API) ListPlugins(instanceID, sleep, timeout int) ([]map[string]interface{}, error) {
return api.listPluginsWithRetry(instanceID, 1, sleep, timeout)
}

func (api *API) readPluginsWithRetry(instanceID, attempts, sleep int) ([]map[string]interface{}, error) {
var data []map[string]interface{}
failed := make(map[string]interface{})
log.Printf("[DEBUG] go-api::plugin::readWithRetry instance id: %v", instanceID)
path := fmt.Sprintf("/api/instances/%d/plugins", instanceID)
response, err := api.sling.New().Get(path).Receive(&data, &failed)
// listPluginsWithRetry: list plugins from an instance, with retry if backend is busy.
func (api *API) listPluginsWithRetry(instanceID, attempt, sleep, timeout int) (
[]map[string]interface{}, error) {

var (
data []map[string]interface{}
failed map[string]interface{}
path = fmt.Sprintf("/api/instances/%d/plugins", instanceID)
)

response, err := api.sling.New().Get(path).Receive(&data, &failed)
if err != nil {
return nil, err
} else if attempt*sleep > timeout {
return nil, fmt.Errorf("read plugins reached timeout of %d seconds", timeout)
}

statusCode := response.StatusCode
log.Printf("[DEBUG] go-api::plugins::readWithRetry statusCode: %d", statusCode)
switch {
case statusCode == 400:
switch response.StatusCode {
case 200:
return data, nil
case 400:
if strings.Compare(failed["error"].(string), "Timeout talking to backend") == 0 {
if attempts--; attempts > 0 {
log.Printf("[INFO] go-api::plugin::readWithRetry Timeout talking to backend "+
"attempts left %d and retry in %d seconds", attempts, sleep)
time.Sleep(time.Duration(sleep) * time.Second)
return api.readPluginsWithRetry(instanceID, attempts, 2*sleep)
} else {
return nil, fmt.Errorf("ReadWithRetry failed, status: %v, message: %s", response.StatusCode, failed)
}
log.Printf("[INFO] go-api::plugins::read Timeout talking to backend "+
"attempt: %d, until timeout: %d", attempt, (timeout - (attempt * sleep)))
attempt++
time.Sleep(time.Duration(sleep) * time.Second)
return api.listPluginsWithRetry(instanceID, attempt, sleep, timeout)
}
return nil, fmt.Errorf("ReadWithRetry failed, status: %v, message: %s", 400, failed)
default:
return nil,
fmt.Errorf("list plugin with retry failed, status: %v, message: %s",
response.StatusCode, failed)
}
return data, nil
}

func (api *API) UpdatePlugin(instanceID int, params map[string]interface{}) (map[string]interface{}, error) {
failed := make(map[string]interface{})
pluginParams := &PluginParams{Name: params["name"].(string), Enabled: params["enabled"].(bool)}
log.Printf("[DEBUG] go-api::plugin::update instance ID: %v, params: %v", instanceID, pluginParams)
path := fmt.Sprintf("/api/instances/%d/plugins?async=true", instanceID)
response, err := api.sling.New().Put(path).BodyJSON(pluginParams).Receive(nil, &failed)
// UpdatePlugin: updates a plugin from an instance.
func (api *API) UpdatePlugin(instanceID int, pluginName string, enabled bool, sleep, timeout int) (
map[string]interface{}, error) {

var (
failed map[string]interface{}
params = make(map[string]interface{})
path = fmt.Sprintf("/api/instances/%d/plugins?async=true", instanceID)
)

params["plugin_name"] = pluginName
params["enabled"] = enabled
log.Printf("[DEBUG] go-api::plugin::update instance ID: %v, params: %v", instanceID, params)
response, err := api.sling.New().Put(path).BodyJSON(params).Receive(nil, &failed)

if err != nil {
return nil, err
}
if response.StatusCode != 204 {
return nil, fmt.Errorf("UpdatePlugin failed, status: %v, message: %s", response.StatusCode, failed)
}

return api.waitUntilPluginChanged(instanceID, params["name"].(string), params["enabled"].(bool))
switch response.StatusCode {
case 204:
return api.waitUntilPluginChanged(instanceID, pluginName, enabled, 1, sleep, timeout)
default:
return nil,
fmt.Errorf("update plugin failed, status: %v, message: %s", response.StatusCode, failed)
}
}

func (api *API) DisablePlugin(instanceID int, pluginName string) (map[string]interface{}, error) {
failed := make(map[string]interface{})
log.Printf("[DEBUG] go-api::plugin::disable instance id: %v, name: %v", instanceID, pluginName)
path := fmt.Sprintf("/api/instances/%d/plugins/%s?async=true", instanceID, pluginName)
// DisablePlugin: disables a plugin from an instance.
func (api *API) DisablePlugin(instanceID int, pluginName string, sleep, timeout int) (
map[string]interface{}, error) {

var (
failed map[string]interface{}
path = fmt.Sprintf("/api/instances/%d/plugins/%s?async=true", instanceID, pluginName)
)

log.Printf("[DEBUG] go-api::plugin::disable path: %s", path)
response, err := api.sling.New().Delete(path).Receive(nil, &failed)

if err != nil {
return nil, err
}
if response.StatusCode != 204 {
return nil, fmt.Errorf("DisablePlugin failed, status: %v, message: %s", response.StatusCode, failed)
}

return api.waitUntilPluginChanged(instanceID, pluginName, false)
switch response.StatusCode {
case 204:
return api.waitUntilPluginChanged(instanceID, pluginName, false, 1, sleep, timeout)
default:
return nil, fmt.Errorf("disable plugin failed, status: %v, message: %s",
response.StatusCode, failed)
}
}

func (api *API) DeletePlugin(instanceID int, pluginName string) error {
failed := make(map[string]interface{})
log.Printf("[DEBUG] go-api::plugin::delete instance: %v, name: %v", instanceID, pluginName)
path := fmt.Sprintf("/api/instances/%d/plugins/%s?async=true", instanceID, pluginName)
response, err := api.sling.New().Delete(path).Receive(nil, &failed)
// DeletePlugin: deletes a plugin from an instance.
func (api *API) DeletePlugin(instanceID int, pluginName string, sleep, timeout int) error {
var (
failed map[string]interface{}
path = fmt.Sprintf("/api/instances/%d/plugins/%s?async=true", instanceID, pluginName)
)

log.Printf("[DEBUG] go-api::plugin::delete path: %s", path)
response, err := api.sling.New().Delete(path).Receive(nil, &failed)
if err != nil {
return err
}
if response.StatusCode != 204 {
return fmt.Errorf("DeletePlugin failed, status: %v, message: %s", response.StatusCode, failed)

switch response.StatusCode {
case 204:
_, err = api.waitUntilPluginChanged(instanceID, pluginName, false, 1, sleep, timeout)
return err
default:
return fmt.Errorf("delete plugin failed, status: %v, message: %s",
response.StatusCode, failed)
}
}

_, err = api.waitUntilPluginChanged(instanceID, pluginName, false)
return err
// waitUntilPluginChanged: wait until plugin changed.
func (api *API) waitUntilPluginChanged(instanceID int, pluginName string, enabled bool,
attempt, sleep, timeout int) (map[string]interface{}, error) {

for {
time.Sleep(time.Duration(sleep) * time.Second)
if attempt*sleep > timeout {
return nil, fmt.Errorf("wait until plugin changed reached timeout of %d seconds", timeout)
}

response, err := api.ReadPlugin(instanceID, pluginName, sleep, timeout)
log.Printf("[DEBUG] go-api::plugin::waitUntilPluginChanged response: %v", response)
if err != nil {
return nil, err
}
if response["required"] != nil && response["required"] != false {
return response, nil
}
if response["enabled"] == enabled {
return response, nil
}
attempt++
}
}
Loading