Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[jaeger-v2] Refactor ElasticSearch/OpenSearch configurations to have more logical groupings #6090

Merged
merged 13 commits into from
Oct 18, 2024
+526 −284
Merged

[jaeger-v2] Refactor ElasticSearch/OpenSearch configurations to have more logical groupings #6090

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
2a47ab5
Add Separate Config Struct For Sniffing
mahadzaryab1 Oct 15, 2024
b8642e7
Group Bulk Processing Configurations Together
mahadzaryab1 Oct 15, 2024
48e652d
Group Authentication Together
mahadzaryab1 Oct 15, 2024
d190973
Add Missing Mapstructure Tags
mahadzaryab1 Oct 15, 2024
dd330c8
Add Documentation To Config Structs
mahadzaryab1 Oct 16, 2024
9b2f124
Use Configured Fields For Initialization
mahadzaryab1 Oct 16, 2024
4aff8fc
Merge branch 'main' into consolidate-es
mahadzaryab1 Oct 16, 2024
c439adf
Merge branch 'main' into consolidate-es
mahadzaryab1 Oct 16, 2024
fa0560b
Address Feedback From PR Review
mahadzaryab1 Oct 16, 2024
7300c33
Create Separate Grouping For Bearer Authentication
mahadzaryab1 Oct 18, 2024
f647de0
Group Fields Together For Readability And Maintainability
mahadzaryab1 Oct 18, 2024
d55411b
Merge branch 'main' into consolidate-es
mahadzaryab1 Oct 18, 2024
dc17b76
Address Feedback From PR Review
mahadzaryab1 Oct 18, 2024
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
192 changes: 130 additions & 62 deletions pkg/es/config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,15 +39,26 @@ const (

// IndexOptions describes the index format and rollover frequency
type IndexOptions struct {
Priority int64 `mapstructure:"priority"`
DateLayout string `mapstructure:"date_layout"`
Shards int64 `mapstructure:"shards"`
Replicas int64 `mapstructure:"replicas"`
// Priority contains the of index template (ESv8 only).
mahadzaryab1 marked this conversation as resolved.
Show resolved Hide resolved
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
Copy link
Member

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?

Copy link
Collaborator Author

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)

Copy link
Member

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.

Copy link
Collaborator Author

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.

Copy link
Member

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))

Copy link
Member

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.

// "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.
Show resolved Hide resolved
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, sett this field to "production" creates "production-jaeger-*".
yurishkuro marked this conversation as resolved.
Show resolved Hide resolved
IndexPrefix IndexPrefix `mapstructure:"index_prefix"`
Spans IndexOptions `mapstructure:"spans"`
Services IndexOptions `mapstructure:"services"`
Expand All @@ -70,34 +81,56 @@ 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.
Show resolved Hide resolved
Servers []string `mapstructure:"server_urls" valid:"required,url"`
RemoteReadClusters []string `mapstructure:"remote_read_clusters"`
Username string `mapstructure:"username"`
Password string `mapstructure:"password" json:"-"`
TokenFilePath string `mapstructure:"token_file"`
PasswordFilePath string `mapstructure:"password_file"`
AllowTokenFromContext bool `mapstructure:"-"`
Sniffer bool `mapstructure:"sniffer"` // https://github.com/olivere/elastic/wiki/Sniffing
SnifferTLSEnabled bool `mapstructure:"sniffer_tls_enabled"`
MaxDocCount int `mapstructure:"-"` // Defines maximum number of results to fetch from storage per query
MaxSpanAge time.Duration `mapstructure:"-"` // configures the maximum lookback on span reads
Timeout time.Duration `mapstructure:"-"`
BulkSize int `mapstructure:"-"`
BulkWorkers int `mapstructure:"-"`
BulkActions int `mapstructure:"-"`
BulkFlushInterval time.Duration `mapstructure:"-"`
Indices Indices `mapstructure:"indices"`
ServiceCacheTTL time.Duration `mapstructure:"service_cache_ttl"`
AdaptiveSamplingLookback time.Duration `mapstructure:"-"`
Tags TagsAsFields `mapstructure:"tags_as_fields"`
Enabled bool `mapstructure:"-"`
TLS configtls.ClientConfig `mapstructure:"tls"`
UseReadWriteAliases bool `mapstructure:"use_aliases"`
CreateIndexTemplates bool `mapstructure:"create_mappings"`
UseILM bool `mapstructure:"use_ilm"`
Version uint `mapstructure:"version"`
LogLevel string `mapstructure:"log_level"`
SendGetBodyAs string `mapstructure:"send_get_body_as"`
// 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"`
// TokenFilePath contains the path to a file containing a bearer token.
// This flag also loads CA if it is specified.
yurishkuro marked this conversation as resolved.
Show resolved Hide resolved
TokenFilePath string `mapstructure:"token_file"`
yurishkuro marked this conversation as resolved.
Show resolved Hide resolved
// AllowTokenFromContext, if set to true, enables reading bearer token from the context.
AllowTokenFromContext bool `mapstructure:"allow_token_from_context"`
Sniffing Sniffing `mapstructure:"sniffing"`
// 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"`
// Timeout contains the timeout used for queries. A timeout of zero means no timeout.
Timeout time.Duration `mapstructure:"timeout"`
BulkProcessing BulkProcessing `mapstructure:"bulk_processing"`
Indices Indices `mapstructure:"indices"`
// 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.
Show resolved Hide resolved
// TLS contains the TLS configuration for the connection to the ElasticSearch clusters.
TLS configtls.ClientConfig `mapstructure:"tls"`
// 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.
Show resolved Hide resolved
UseILM bool `mapstructure:"use_ilm"`
// 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"`
// SendGetBodyAs is the HTTP verb to use for requests that contain a body.
SendGetBodyAs string `mapstructure:"send_get_body_as"`
}

// TagsAsFields holds configuration for tag schema.
Expand All @@ -114,6 +147,41 @@ 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 {
// Size, contains the number of bytes which specifies when to flush.
Size int `mapstructure:"size"`
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"`
// Actions contain the number of added actions which specifies when to flush.
Actions int `mapstructure:"actions"`
mahadzaryab1 marked this conversation as resolved.
Show resolved Hide resolved
// 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
}

type Authentication struct {
BasicAuthentication BasicAuthentication `mapstructure:"basic"`
}

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"`
}

// NewClient creates a new ElasticSearch client
func NewClient(c *Configuration, logger *zap.Logger, metricsFactory metrics.Factory) (es.Client, error) {
if len(c.Servers) < 1 {
Expand Down Expand Up @@ -171,10 +239,10 @@ func NewClient(c *Configuration, logger *zap.Logger, metricsFactory metrics.Fact
zap.Any("response", response))
}
}).
BulkSize(c.BulkSize).
Workers(c.BulkWorkers).
BulkActions(c.BulkActions).
FlushInterval(c.BulkFlushInterval).
BulkSize(c.BulkProcessing.Size).
Workers(c.BulkProcessing.Workers).
BulkActions(c.BulkProcessing.Actions).
FlushInterval(c.BulkProcessing.FlushInterval).
Do(context.Background())
if err != nil {
return nil, err
Expand Down Expand Up @@ -220,9 +288,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.Username
options.Password = c.Password
options.DiscoverNodesOnStart = c.Sniffer
options.Username = c.Authentication.BasicAuthentication.Username
options.Password = c.Authentication.BasicAuthentication.Password
options.DiscoverNodesOnStart = c.Sniffing.Enabled
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

so for v8 only the enabled is used, not the UseHTTPS? Is it because v8 client always uses https?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@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://"
...
}

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would be good to have a comment on the field mentioning that.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yurishkuro #6103

transport, err := GetHTTPRoundTripper(c, logger)
if err != nil {
return nil, err
Expand Down Expand Up @@ -258,14 +326,14 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
if len(c.RemoteReadClusters) == 0 {
c.RemoteReadClusters = source.RemoteReadClusters
}
if c.Username == "" {
c.Username = source.Username
if c.Authentication.BasicAuthentication.Username == "" {
c.Authentication.BasicAuthentication.Username = source.Authentication.BasicAuthentication.Username
}
if c.Password == "" {
c.Password = source.Password
if c.Authentication.BasicAuthentication.Password == "" {
c.Authentication.BasicAuthentication.Password = source.Authentication.BasicAuthentication.Password
}
if !c.Sniffer {
c.Sniffer = source.Sniffer
if !c.Sniffing.Enabled {
c.Sniffing.Enabled = source.Sniffing.Enabled
}
if c.MaxSpanAge == 0 {
c.MaxSpanAge = source.MaxSpanAge
Expand All @@ -281,20 +349,20 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
setDefaultIndexOptions(&c.Indices.Services, &source.Indices.Services)
setDefaultIndexOptions(&c.Indices.Dependencies, &source.Indices.Dependencies)

if c.BulkSize == 0 {
c.BulkSize = source.BulkSize
if c.BulkProcessing.Size == 0 {
c.BulkProcessing.Size = source.BulkProcessing.Size
}
if c.BulkWorkers == 0 {
c.BulkWorkers = source.BulkWorkers
if c.BulkProcessing.Workers == 0 {
c.BulkProcessing.Workers = source.BulkProcessing.Workers
}
if c.BulkActions == 0 {
c.BulkActions = source.BulkActions
if c.BulkProcessing.Actions == 0 {
c.BulkProcessing.Actions = source.BulkProcessing.Actions
}
if c.BulkFlushInterval == 0 {
c.BulkFlushInterval = source.BulkFlushInterval
if c.BulkProcessing.FlushInterval == 0 {
c.BulkProcessing.FlushInterval = source.BulkProcessing.FlushInterval
}
if !c.SnifferTLSEnabled {
c.SnifferTLSEnabled = source.SnifferTLSEnabled
if !c.Sniffing.UseHTTPS {
c.Sniffing.UseHTTPS = source.Sniffing.UseHTTPS
}
if !c.Tags.AllAsFields {
c.Tags.AllAsFields = source.Tags.AllAsFields
Expand Down Expand Up @@ -361,31 +429,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.Sniffer),
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' 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.AllowTokenFromContext),
}
if c.SnifferTLSEnabled {
if c.Sniffing.UseHTTPS {
options = append(options, elastic.SetScheme("https"))
}
httpClient := &http.Client{
Timeout: c.Timeout,
}
options = append(options, elastic.SetHttpClient(httpClient))

if c.Password != "" && c.PasswordFilePath != "" {
if c.Authentication.BasicAuthentication.Password != "" && c.Authentication.BasicAuthentication.PasswordFilePath != "" {
return nil, fmt.Errorf("both Password and PasswordFilePath are set")
}
if c.PasswordFilePath != "" {
passwordFromFile, err := loadTokenFromFile(c.PasswordFilePath)
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.Password = passwordFromFile
c.Authentication.BasicAuthentication.Password = passwordFromFile
}
options = append(options, elastic.SetBasicAuth(c.Username, c.Password))
options = append(options, elastic.SetBasicAuth(c.Authentication.BasicAuthentication.Username, c.Authentication.BasicAuthentication.Password))

if c.SendGetBodyAs != "" {
options = append(options, elastic.SetSendGetBodyAs(c.SendGetBodyAs))
Expand Down
Loading
Loading