[jaeger-v2] Refactor ElasticSearch/OpenSearch configurations to have more logical groupings #6090
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 |
|---|---|---|
|
|
@@ -39,15 +39,25 @@ const ( | |
|
|
||
| // IndexOptions describes the index format and rollover frequency | ||
| type IndexOptions struct { | ||
| // Priority contains the priority of index template (ESv8 only). | ||
| Priority int64 `mapstructure:"priority"` | ||
| DateLayout string `mapstructure:"date_layout"` | ||
| // Shards is the number of shards per index in Elasticsearch. | ||
| Shards int64 `mapstructure:"shards"` | ||
| // Replicas is the number of replicas per index in Elasticsearch. | ||
| Replicas int64 `mapstructure:"replicas"` | ||
| // RolloverFrequency contains the rollover frequency setting used to fetch | ||
| // indices from elasticsearch. | ||
| // Valid configuration options are: [hour, day]. | ||
| // This setting does not affect the index rotation and is simply used for | ||
| // fetching indices. | ||
| RolloverFrequency string `mapstructure:"rollover_frequency"` | ||
| } | ||
|
|
||
| // Indices describes different configuration options for each index type | ||
| type Indices struct { | ||
| // IndexPrefix is an optional prefix to prepend to Jaeger indices. | ||
| // For example, setting this field to "production" creates "production-jaeger-*". | ||
| IndexPrefix IndexPrefix `mapstructure:"index_prefix"` | ||
| Spans IndexOptions `mapstructure:"spans"` | ||
| Services IndexOptions `mapstructure:"services"` | ||
|
|
@@ -70,34 +80,58 @@ func (p IndexPrefix) Apply(indexName string) string { | |
|
|
||
| // Configuration describes the configuration properties needed to connect to an ElasticSearch cluster | ||
| type Configuration struct { | ||
| // ---- connection related configs ---- | ||
| // Servers is a list of Elasticsearch servers. The strings must must contain full URLs | ||
| // (i.e. http://localhost:9200). | ||
| Servers []string `mapstructure:"server_urls" valid:"required,url"` | ||
| // RemoteReadClusters is a list of Elasticsearch remote cluster names for cross-cluster | ||
| // querying. | ||
| RemoteReadClusters []string `mapstructure:"remote_read_clusters"` | ||
| Authentication Authentication `mapstructure:"auth"` | ||
| // TLS contains the TLS configuration for the connection to the ElasticSearch clusters. | ||
| TLS configtls.ClientConfig `mapstructure:"tls"` | ||
| Sniffing Sniffing `mapstructure:"sniffing"` | ||
| // SendGetBodyAs is the HTTP verb to use for requests that contain a body. | ||
| SendGetBodyAs string `mapstructure:"send_get_body_as"` | ||
| // QueryTimeout contains the timeout used for queries. A timeout of zero means no timeout. | ||
| QueryTimeout time.Duration `mapstructure:"query_timeout"` | ||
|
|
||
| // ---- elasticsearch client related configs ---- | ||
| BulkProcessing BulkProcessing `mapstructure:"bulk_processing"` | ||
| // Version contains the major Elasticsearch version. If this field is not specified, | ||
| // the value will be auto-detected from Elasticsearch. | ||
| Version uint `mapstructure:"version"` | ||
| // LogLevel contains the Elasticsearch client log-level. Valid values for this field | ||
| // are: [debug, info, error] | ||
| LogLevel string `mapstructure:"log_level"` | ||
|
|
||
| // ---- index related configs ---- | ||
| Indices Indices `mapstructure:"indices"` | ||
| // UseReadWriteAliases, if set to true, will use read and write aliases for indices. | ||
| // Use this option with Elasticsearch rollover API. It requires an external component | ||
| // to create aliases before startup and then performing its management. | ||
| UseReadWriteAliases bool `mapstructure:"use_aliases"` | ||
| // CreateIndexTemplates, if set to true, creates index templates at application startup. | ||
| // This configuration should be set to false when templates are installed manually. | ||
| CreateIndexTemplates bool `mapstructure:"create_mappings"` | ||
| // Option to enable Index Lifecycle Management (ILM) for Jaeger span and service indices. | ||
| // Read more about ILM at | ||
| // https://www.jaegertracing.io/docs/deployment/#enabling-ilm-support | ||
| UseILM bool `mapstructure:"use_ilm"` | ||
|
|
||
| // ---- jaeger-specific configs ---- | ||
| // MaxDocCount Defines maximum number of results to fetch from storage per query. | ||
| MaxDocCount int `mapstructure:"max_doc_count"` | ||
| // MaxSpanAge configures the maximum lookback on span reads. | ||
| MaxSpanAge time.Duration `mapstructure:"max_span_age"` | ||
| // ServiceCacheTTL contains the TTL for the cache of known service names. | ||
| ServiceCacheTTL time.Duration `mapstructure:"service_cache_ttl"` | ||
| // AdaptiveSamplingLookback contains the duration to look back for the | ||
| // latest adaptive sampling probabilities. | ||
| AdaptiveSamplingLookback time.Duration `mapstructure:"adaptive_sampling_lookback"` | ||
| Tags TagsAsFields `mapstructure:"tags_as_fields"` | ||
| // Enabled, if set to true, enables the namespace for storage pointed to by this configuration. | ||
| Enabled bool `mapstructure:"-"` | ||
| } | ||
|
|
||
| // TagsAsFields holds configuration for tag schema. | ||
|
|
@@ -114,6 +148,53 @@ type TagsAsFields struct { | |
| Include string `mapstructure:"include"` | ||
| } | ||
|
|
||
| // Sniffing sets the sniffing configuration for the ElasticSearch client, which is the process | ||
| // of finding all the nodes of your cluster. Read more about sniffing at | ||
| // https://github.com/olivere/elastic/wiki/Sniffing. | ||
| type Sniffing struct { | ||
| // Enabled, if set to true, enables sniffing for the ElasticSearch client. | ||
| Enabled bool `mapstructure:"enabled"` | ||
| // UseHTTPS, if set to true, sets the HTTP scheme to HTTPS when performing sniffing. | ||
| UseHTTPS bool `mapstructure:"use_https"` | ||
| } | ||
|
|
||
| type BulkProcessing struct { | ||
| // MaxBytes, contains the number of bytes which specifies when to flush. | ||
| MaxBytes int `mapstructure:"max_bytes"` | ||
| // MaxActions contain the number of added actions which specifies when to flush. | ||
| MaxActions int `mapstructure:"max_actions"` | ||
| // FlushInterval is the interval at the end of which a flush occurs. | ||
| FlushInterval time.Duration `mapstructure:"flush_interval"` | ||
mahadzaryab1 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| // Workers contains the number of concurrent workers allowed to be executed. | ||
| Workers int `mapstructure:"workers"` | ||
| } | ||
|
|
||
| type Authentication struct { | ||
| BasicAuthentication BasicAuthentication `mapstructure:"basic"` | ||
| BearerTokenAuthentication BearerTokenAuthentication `mapstructure:"bearer_token"` | ||
| } | ||
|
|
||
| type BasicAuthentication struct { | ||
| // Username contains the username required to connect to Elasticsearch. | ||
| Username string `mapstructure:"username"` | ||
| // Password contains The password required by Elasticsearch | ||
| Password string `mapstructure:"password" json:"-"` | ||
| // PasswordFilePath contains the path to a file containing password. | ||
| // This file is watched for changes. | ||
| PasswordFilePath string `mapstructure:"password_file"` | ||
| } | ||
|
|
||
| // BearerTokenAuthentication contains the configuration for attaching bearer tokens | ||
| // when making HTTP requests. Note that TokenFilePath and AllowTokenFromContext | ||
| // should not both be enabled. If both TokenFilePath and AllowTokenFromContext are set, | ||
| // the TokenFilePath will be ignored. | ||
| type BearerTokenAuthentication struct { | ||
| // FilePath contains the path to a file containing a bearer token. | ||
| FilePath string `mapstructure:"file_path"` | ||
| // AllowTokenFromContext, if set to true, enables reading bearer token from the context. | ||
| AllowFromContext bool `mapstructure:"from_context"` | ||
| } | ||
|
|
||
| // NewClient creates a new ElasticSearch client | ||
| func NewClient(c *Configuration, logger *zap.Logger, metricsFactory metrics.Factory) (es.Client, error) { | ||
| if len(c.Servers) < 1 { | ||
|
|
@@ -171,10 +252,10 @@ func NewClient(c *Configuration, logger *zap.Logger, metricsFactory metrics.Fact | |
| zap.Any("response", response)) | ||
| } | ||
| }). | ||
| BulkSize(c.BulkProcessing.MaxBytes). | ||
| Workers(c.BulkProcessing.Workers). | ||
| BulkActions(c.BulkProcessing.MaxActions). | ||
| FlushInterval(c.BulkProcessing.FlushInterval). | ||
| Do(context.Background()) | ||
| if err != nil { | ||
| return nil, err | ||
|
|
@@ -220,9 +301,9 @@ func NewClient(c *Configuration, logger *zap.Logger, metricsFactory metrics.Fact | |
| func newElasticsearchV8(c *Configuration, logger *zap.Logger) (*esV8.Client, error) { | ||
| var options esV8.Config | ||
| options.Addresses = c.Servers | ||
| options.Username = c.Authentication.BasicAuthentication.Username | ||
| options.Password = c.Authentication.BasicAuthentication.Password | ||
| options.DiscoverNodesOnStart = c.Sniffing.Enabled | ||
|
There was a problem hiding this comment. so for v8 only the There was a problem hiding this comment. @yurishkuro yeah, that seems to be the case. The following is used when constructing the client. // addrFromCloudID extracts the Elasticsearch URL from CloudID.
// See: https://www.elastic.co/guide/en/cloud/current/ec-cloud-id.html
func addrFromCloudID(input string) (string, error) {
var scheme = "https://"
...
}There was a problem hiding this comment. It would be good to have a comment on the field mentioning that. There was a problem hiding this comment. |
||
| transport, err := GetHTTPRoundTripper(c, logger) | ||
| if err != nil { | ||
| return nil, err | ||
|
|
@@ -258,14 +339,14 @@ func (c *Configuration) ApplyDefaults(source *Configuration) { | |
| if len(c.RemoteReadClusters) == 0 { | ||
| c.RemoteReadClusters = source.RemoteReadClusters | ||
| } | ||
| if c.Authentication.BasicAuthentication.Username == "" { | ||
| c.Authentication.BasicAuthentication.Username = source.Authentication.BasicAuthentication.Username | ||
| } | ||
| if c.Authentication.BasicAuthentication.Password == "" { | ||
| c.Authentication.BasicAuthentication.Password = source.Authentication.BasicAuthentication.Password | ||
| } | ||
| if !c.Sniffing.Enabled { | ||
| c.Sniffing.Enabled = source.Sniffing.Enabled | ||
| } | ||
| if c.MaxSpanAge == 0 { | ||
| c.MaxSpanAge = source.MaxSpanAge | ||
|
|
@@ -281,20 +362,20 @@ func (c *Configuration) ApplyDefaults(source *Configuration) { | |
| setDefaultIndexOptions(&c.Indices.Services, &source.Indices.Services) | ||
| setDefaultIndexOptions(&c.Indices.Dependencies, &source.Indices.Dependencies) | ||
|
|
||
| if c.BulkProcessing.MaxBytes == 0 { | ||
| c.BulkProcessing.MaxBytes = source.BulkProcessing.MaxBytes | ||
| } | ||
| if c.BulkProcessing.Workers == 0 { | ||
| c.BulkProcessing.Workers = source.BulkProcessing.Workers | ||
| } | ||
| if c.BulkProcessing.MaxActions == 0 { | ||
| c.BulkProcessing.MaxActions = source.BulkProcessing.MaxActions | ||
| } | ||
| if c.BulkProcessing.FlushInterval == 0 { | ||
| c.BulkProcessing.FlushInterval = source.BulkProcessing.FlushInterval | ||
| } | ||
| if !c.Sniffing.UseHTTPS { | ||
| c.Sniffing.UseHTTPS = source.Sniffing.UseHTTPS | ||
| } | ||
| if !c.Tags.AllAsFields { | ||
| c.Tags.AllAsFields = source.Tags.AllAsFields | ||
|
|
@@ -361,31 +442,31 @@ func (c *Configuration) TagKeysAsFields() ([]string, error) { | |
| // getConfigOptions wraps the configs to feed to the ElasticSearch client init | ||
| func (c *Configuration) getConfigOptions(logger *zap.Logger) ([]elastic.ClientOptionFunc, error) { | ||
| options := []elastic.ClientOptionFunc{ | ||
| elastic.SetURL(c.Servers...), elastic.SetSniff(c.Sniffing.Enabled), | ||
| // Disable health check when token from context is allowed, this is because at this time | ||
| // we don'r have a valid token to do the check ad if we don't disable the check the service that | ||
| // uses this won't start. | ||
| elastic.SetHealthcheck(!c.Authentication.BearerTokenAuthentication.AllowFromContext), | ||
| } | ||
| if c.Sniffing.UseHTTPS { | ||
| options = append(options, elastic.SetScheme("https")) | ||
| } | ||
| httpClient := &http.Client{ | ||
| Timeout: c.QueryTimeout, | ||
| } | ||
| options = append(options, elastic.SetHttpClient(httpClient)) | ||
|
|
||
| if c.Authentication.BasicAuthentication.Password != "" && c.Authentication.BasicAuthentication.PasswordFilePath != "" { | ||
| return nil, fmt.Errorf("both Password and PasswordFilePath are set") | ||
| } | ||
| if c.Authentication.BasicAuthentication.PasswordFilePath != "" { | ||
| passwordFromFile, err := loadTokenFromFile(c.Authentication.BasicAuthentication.PasswordFilePath) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to load password from file: %w", err) | ||
| } | ||
| c.Authentication.BasicAuthentication.Password = passwordFromFile | ||
| } | ||
| options = append(options, elastic.SetBasicAuth(c.Authentication.BasicAuthentication.Username, c.Authentication.BasicAuthentication.Password)) | ||
|
|
||
| if c.SendGetBodyAs != "" { | ||
| options = append(options, elastic.SetSendGetBodyAs(c.SendGetBodyAs)) | ||
|
|
@@ -465,20 +546,20 @@ func GetHTTPRoundTripper(c *Configuration, logger *zap.Logger) (http.RoundTrippe | |
| } | ||
|
|
||
| token := "" | ||
| if c.Authentication.BearerTokenAuthentication.FilePath != "" { | ||
| if c.Authentication.BearerTokenAuthentication.AllowFromContext { | ||
| logger.Warn("Token file and token propagation are both enabled, token from file won't be used") | ||
| } | ||
| tokenFromFile, err := loadTokenFromFile(c.Authentication.BearerTokenAuthentication.FilePath) | ||
|
There was a problem hiding this comment. weird that despite the PR title #4342 there is no "reloading" of the file, only at init time (whereas password does get reloaded). But since nobody complained ... I don't see much documentation on the website about how this token is supposed to work, specifically any pointers to the documentation for the actual storage (e.g. ES or hosted ES). Might have been some references in the original issues but hard to find them. There was a problem hiding this comment. |
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| token = tokenFromFile | ||
| } | ||
| if token != "" || c.Authentication.BearerTokenAuthentication.AllowFromContext { | ||
| transport = bearertoken.RoundTripper{ | ||
| Transport: httpTransport, | ||
| OverrideFromCtx: c.Authentication.BearerTokenAuthentication.AllowFromContext, | ||
| StaticToken: token, | ||
| } | ||
| } | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.