Make proxy optional for k8s_metadata_decorator (#239)

* Added `allow_proxy` option (default: false) for enabling or
disabling the use of a proxy for this operator. Generally, you would
not want to use a proxy for connections to the Kubernetes API.

The Kubernetes client, by default, will use any proxy set at HTTP_PROXY or
HTTPS_PROXY unless the proxy function is set.

supporting docs:
- https://godoc.org/k8s.io/client-go/rest#Config
- https://golang.org/pkg/net/http/#ProxyFromEnvironment

resolves #238

* changelog: using a proxy causes internal API timeout

* Clarify that boolean is enabling / disabling proxy support when communicating with the k8s api

* Use mocked time function for tests

Co-authored-by: Dan Jaglowski <[email protected]>

CHANGELOG.md

@@ -3,6 +3,11 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## Unreleased
+
+### Fixed
+- `k8s_metadata_decorator` using a proxy causes internal API timeout
+
 ## [0.13.8] - 2020-12-30
 ### Fixed
 - `file_input` exclude processing could result in extra exclusions

docs/operators/k8s_metadata_decorator.md

@@ -12,6 +12,7 @@ The `k8s_metadata_decorator` operator adds labels and annotations to the entry u
 | `pod_name_field`  | `pod_name`               | A [field](/docs/types/field.md) that contains the k8s pod name associated with the log entry                                                                                                                                             |
 | `cache_ttl`       | 10m                      | A [duration](/docs/types/duration.md) indicating the time it takes for a cached entry to expire                                                                                                                                          |
 | `timeout`         | 10s                      | A [duration](/docs/types/duration.md) indicating how long to wait for the API to respond before timing out                                                                                                                               |
+| `allow_proxy`     | false                    | Controls whether or not the agent will take into account [proxy](https://github.com/observIQ/stanza/blob/master/docs/proxy.md) configuration when communicating with the k8s metadata api |
 | `if`              |                          | An [expression](/docs/types/expression.md) that, when set, will be evaluated to determine whether this operator should be used for the given entry. This allows you to do easy conditional parsing without branching logic with routers. |
 
 ### Example Configurations

operator/builtin/parser/syslog/syslog_test.go

@@ -12,6 +12,10 @@ import (
 )
 
 func TestSyslogParser(t *testing.T) {
+	now := func() time.Time {
+		return time.Date(2020, 06, 16, 3, 31, 34, 525, time.UTC)
+	}
+
 	basicConfig := func() *SyslogParserConfig {
 		cfg := NewSyslogParserConfig("test_operator_id")
 		cfg.OutputIDs = []string{"fake"}
@@ -35,7 +39,7 @@ func TestSyslogParser(t *testing.T) {
 				return cfg
 			}(),
 			"<34>Jan 12 06:30:00 1.2.3.4 apache_server: test message",
-			time.Date(time.Now().Year(), 1, 12, 6, 30, 0, 0, time.UTC),
+			time.Date(now().Year(), 1, 12, 6, 30, 0, 0, time.UTC),
 			map[string]interface{}{
 				"appname":  "apache_server",
 				"facility": 4,
@@ -54,7 +58,7 @@ func TestSyslogParser(t *testing.T) {
 				return cfg
 			}(),
 			[]byte("<34>Jan 12 06:30:00 1.2.3.4 apache_server: test message"),
-			time.Date(time.Now().Year(), 1, 12, 6, 30, 0, 0, time.UTC),
+			time.Date(now().Year(), 1, 12, 6, 30, 0, 0, time.UTC),
 			map[string]interface{}{
 				"appname":  "apache_server",
 				"facility": 4,

operator/builtin/transformer/k8smetadata/k8s_metadata_decorator.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"sync"
 	"time"
+	"net/url"
+	"net/http"
 
 	"github.com/observiq/stanza/entry"
 	"github.com/observiq/stanza/errors"
@@ -29,6 +31,7 @@ func NewK8sMetadataDecoratorConfig(operatorID string) *K8sMetadataDecoratorConfi
 		NamespaceField:    entry.NewResourceField("k8s.namespace.name"),
 		CacheTTL:          helper.Duration{Duration: 10 * time.Minute},
 		Timeout:           helper.Duration{Duration: 10 * time.Second},
+		AllowProxy:        false,
 	}
 }
 
@@ -39,6 +42,7 @@ type K8sMetadataDecoratorConfig struct {
 	NamespaceField           entry.Field     `json:"namespace_field,omitempty" yaml:"namespace_field,omitempty"`
 	CacheTTL                 helper.Duration `json:"cache_ttl,omitempty"       yaml:"cache_ttl,omitempty"`
 	Timeout                  helper.Duration `json:"timeout,omitempty"         yaml:"timeout,omitempty"`
+	AllowProxy               bool            `json:"allow_proxy,omitempty"     yaml:"allow_proxy,omitempty"`
 }
 
 // Build will build a k8s_metadata_decorator operator from the supplied configuration
@@ -54,6 +58,7 @@ func (c K8sMetadataDecoratorConfig) Build(context operator.BuildContext) ([]oper
 		namespaceField:      c.NamespaceField,
 		cacheTTL:            c.CacheTTL.Raw(),
 		timeout:             c.Timeout.Raw(),
+		allowProxy:          c.AllowProxy,
 	}
 
 	return []operator.Operator{op}, nil
@@ -73,6 +78,7 @@ type K8sMetadataDecorator struct {
 	podCache       MetadataCache
 	cacheTTL       time.Duration
 	timeout        time.Duration
+	allowProxy     bool
 }
 
 // MetadataCacheEntry is an entry in the metadata cache
@@ -115,6 +121,12 @@ func (k *K8sMetadataDecorator) Start() error {
 		)
 	}
 
+	if ! k.allowProxy {
+		config.Proxy = func (*http.Request) (*url.URL, error) {
+			return nil, nil
+		}
+	}
+
 	clientset, err := kubernetes.NewForConfig(config)
 	if err != nil {
 		return errors.Wrap(err, "build client set")

operator/builtin/transformer/k8smetadata/k8s_metadata_decorator_test.go

@@ -54,6 +54,7 @@ func TestK8sMetadataDecoratorBuildDefault(t *testing.T) {
 		namespaceField: entry.NewResourceField("k8s.namespace.name"),
 		cacheTTL:       10 * time.Minute,
 		timeout:        10 * time.Second,
+		allowProxy:     false,
 	}
 
 	ops, err := cfg.Build(testutil.NewBuildContext(t))