Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
# REQUIRED
# Kind can be one of:
# - breaking-change: a change to previously-documented behavior
# - deprecation: functionality that is being removed in a later release
# - bug-fix: fixes a problem in a previous version
# - enhancement: extends functionality but does not break or fix existing behavior
# - feature: new functionality
# - known-issue: problems that we are aware of in a given version
# - security: impacts on the security of a product or a user’s deployment.
# - upgrade: important information for someone upgrading from a prior version
# - other: does not fit into any of the other categories
kind: bug-fix

# REQUIRED for all kinds
# Change summary; a 80ish characters long description of the change.
summary: autoops agent to shutdown when it can't recover from the http errors

# REQUIRED for breaking-change, deprecation, known-issue
# Long description; in case the summary is not enough to describe the change
# this field accommodate a description without length limits.
# description:

# REQUIRED for breaking-change, deprecation, known-issue
# impact:

# REQUIRED for breaking-change, deprecation, known-issue
# action:

# REQUIRED for all kinds
# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
component: metricbeat

# AUTOMATED
# OPTIONAL to manually add other PR URLs
# PR URL: A link the PR that added the changeset.
# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
# Please provide it if you are adding a fragment for a different PR.
# pr: https://github.com/owner/repo/1234

# AUTOMATED
# OPTIONAL to manually add other issue URLs
# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
# If not present is automatically filled by the tooling with the issue linked to the PR number.
# issue: https://github.com/owner/repo/1234
40 changes: 27 additions & 13 deletions x-pack/metricbeat/module/autoops_es/metricset/cluster_info.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,12 @@
package metricset

import (
"os"
"time"
"errors"
"fmt"
"slices"

"github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
"github.com/elastic/beats/v7/x-pack/metricbeat/module/autoops_es/utils"
"github.com/elastic/elastic-agent-libs/logp"
libversion "github.com/elastic/elastic-agent-libs/version"
)

Expand All @@ -19,10 +19,20 @@ const MinimumEsVersion = "7.17.0"
var minVersion = libversion.MustNew(MinimumEsVersion)
var isVersionChecked = false

const (
CLUSTER_INFO_INITIAL_ERROR = 1
CLUSTER_INFO_RUNTIME_ERROR = 2
)

// List of HTTP status codes that indicate the agent cannot recover
var terminalHttpErrorStatusCodes = []int{401, 403, 404}

func GetInfo(m *elasticsearch.MetricSet) (*utils.ClusterInfo, error) {
info, err := utils.FetchAPIData[utils.ClusterInfo](m, "/")

if err != nil {
var httpResponse *utils.HTTPResponse
handleClusterInfoError(m, err, httpResponse)
return nil, err
} else if info.ClusterID == "" || info.ClusterID == "_na_" {
return nil, &utils.ClusterInfoError{Message: "cluster ID is unset, which means the cluster is not ready"}
Expand All @@ -32,7 +42,7 @@ func GetInfo(m *elasticsearch.MetricSet) (*utils.ClusterInfo, error) {
if !isVersionChecked {
// for some reason log.Fatal() isn't working properly so we need to handle the error in a goroutine
errChan := make(chan error)
go handleErrors(m.Logger(), errChan)
go handleFatalErrors(m.Logger(), errChan, CLUSTER_INFO_INITIAL_ERROR)

if err := checkEsVersion(info.Version.Number, errChan); err != nil {
return nil, err
Expand All @@ -42,6 +52,19 @@ func GetInfo(m *elasticsearch.MetricSet) (*utils.ClusterInfo, error) {
return info, nil
}

func handleClusterInfoError(m *elasticsearch.MetricSet, err error, httpResponse *utils.HTTPResponse) {
if errors.As(err, &httpResponse) {
if slices.Contains(terminalHttpErrorStatusCodes, httpResponse.StatusCode) {
// in these error cases Autoops agent can't recover itself, hence stop the agent
errChan := make(chan error)
go handleFatalErrors(m.Logger(), errChan, CLUSTER_INFO_RUNTIME_ERROR)
customErr := fmt.Errorf("autoops agent can't fetch the metrics due to http error! Code: %d, Status: %s",
httpResponse.StatusCode, httpResponse.Status)
errChan <- customErr
}
}
}

func checkEsVersion(esVersion *libversion.V, errChan chan error) error {
if esVersion.LessThan(minVersion) {
isVersionChecked = true
Expand All @@ -55,12 +78,3 @@ func checkEsVersion(esVersion *libversion.V, errChan chan error) error {

return nil
}

func handleErrors(logger *logp.Logger, errChan chan error) {
for err := range errChan {
logger.Error(err)
// sleep is needed to make sure the error is logged and error event is sent before exiting
time.Sleep(time.Second * 5)
os.Exit(1)
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
// or more contributor license agreements. Licensed under the Elastic License;
// you may not use this file except in compliance with the Elastic License.

package metricset

import (
"os"
"time"

"github.com/elastic/elastic-agent-libs/logp"
)

func handleFatalErrors(logger *logp.Logger, errChan chan error, errorCode int) {
for err := range errChan {
logger.Error(err)
// sleep is needed to make sure the error is logged and error event is sent before exiting
time.Sleep(time.Second * 5)
os.Exit(errorCode)
}
}
Loading