[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,26 @@ 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 rotates indices over the given period. For example, "day" creates | ||
| // "jaeger-{index_type}-yyyy-MM-dd" every day after UTC 12AM. | ||
| // Valid configuration options are: [hour, day]. | ||
| // This does not delete old indices. | ||
| // For details on complete index management solutions supported by Jaeger, | ||
| // refer to: https://www.jaegertracing.io/docs/deployment/#elasticsearch-rollover | ||
|
yurishkuro marked this conversation as resolved.
Outdated
|
||
| RolloverFrequency string `mapstructure:"rollover_frequency"` // "hour" or "day" | ||
| } | ||
|
|
||
| // 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 +81,54 @@ func (p IndexPrefix) Apply(indexName string) string { | |
|
|
||
| // Configuration describes the configuration properties needed to connect to an ElasticSearch cluster | ||
| type Configuration struct { | ||
|
yurishkuro marked this conversation as resolved.
|
||
| // 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"` | ||
| // Timeout contains the timeout used for queries. A timeout of zero means no timeout. | ||
| Timeout time.Duration `mapstructure:"timeout"` | ||
|
yurishkuro marked this conversation as resolved.
Outdated
|
||
|
|
||
| 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"` | ||
|
|
||
| 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.elastic.co/guide/en/elasticsearch/reference/current/index-lifecycle-management.html. | ||
|
yurishkuro marked this conversation as resolved.
Outdated
|
||
| UseILM bool `mapstructure:"use_ilm"` | ||
|
|
||
| // 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:"enable"` | ||
|
yurishkuro marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
||
| // TagsAsFields holds configuration for tag schema. | ||
|
|
@@ -114,6 +145,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.
|
||
| // Workers contains the number of concurrent workers allowed to be executed. | ||
| Workers int `mapstructure:"workers"` | ||
| } | ||
|
|
||
| type Authentication struct { | ||
| BasicAuthentication BasicAuthentication `mapstructure:"basic"` | ||
| BearerAuthentication BearerAuthentication `mapstructure:"bearer"` | ||
|
yurishkuro marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
||
| 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"` | ||
| } | ||
|
|
||
| // BearerAuthentication 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 BearerAuthentication 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. | ||
| FromContext bool `mapstructure:"from_context"` | ||
|
yurishkuro marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
||
| // 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 +249,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 +298,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 | ||
|
Member
There was a problem hiding this comment. so for v8 only the
Collaborator
Author
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://"
...
}
Member
There was a problem hiding this comment. It would be good to have a comment on the field mentioning that.
Collaborator
Author
There was a problem hiding this comment. |
||
| transport, err := GetHTTPRoundTripper(c, logger) | ||
| if err != nil { | ||
| return nil, err | ||
|
|
@@ -258,14 +336,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 +359,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 +439,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.BearerAuthentication.FromContext), | ||
| } | ||
| if c.Sniffing.UseHTTPS { | ||
| options = append(options, elastic.SetScheme("https")) | ||
| } | ||
| httpClient := &http.Client{ | ||
| Timeout: c.Timeout, | ||
| } | ||
| 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 +543,20 @@ func GetHTTPRoundTripper(c *Configuration, logger *zap.Logger) (http.RoundTrippe | |
| } | ||
|
|
||
| token := "" | ||
| if c.Authentication.BearerAuthentication.FilePath != "" { | ||
| if c.Authentication.BearerAuthentication.FromContext { | ||
| logger.Warn("Token file and token propagation are both enabled, token from file won't be used") | ||
| } | ||
| tokenFromFile, err := loadTokenFromFile(c.Authentication.BearerAuthentication.FilePath) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| token = tokenFromFile | ||
| } | ||
| if token != "" || c.Authentication.BearerAuthentication.FromContext { | ||
| transport = bearertoken.RoundTripper{ | ||
| Transport: httpTransport, | ||
| OverrideFromCtx: c.Authentication.BearerAuthentication.FromContext, | ||
| StaticToken: token, | ||
| } | ||
| } | ||
|
|
||
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for adding docs. Is this behavior unconditional or only in some cases, like when useILM is false?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@yurishkuro I believe its unconditional. it seems to be used in the reader (https://github.com/jaegertracing/jaeger/blob/main/plugin/storage/es/spanstore/reader.go#L275)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
no, it's not unconditional. See reader.go:151 - this param is not used when
archive || useReadWriteAliases. Also, the doc string is not even talking about using the param at read time, it says "rotates" - and I think that's definitely not unconditional.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@yurishkuro Is this rotation handled by ILM and this field here is just used for reading? https://github.com/jaegertracing/jaeger/blob/main/plugin/storage/es/factory.go#L273-L282. I don't see this field being used for any writing.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, I think you were right. The index rotation is always external to Jaeger, either via a rollover script or via ILM. And the reader only needs this setting to know how to translate a [from, to] interval into distinct index names (by doing
t=t.Add(interval))There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
But given that, the doc string is wrong as it talks about creating rollover indices.