httpprovider for Collector: load configuration from config files in H…

…TTP servers (#5876)

* httpprovider to upstream

* no slashes in the prefix check

* use httptest package to do unit tests; then add one more testcase TestValidateProviderScheme

* added error checks for http.ResponseWriter.Write()

* Fixed bug of variable shadows, then replace old io/ioutil package with io and os

* update the old name ConfigMapProvider

* Remove unnecessary interface of httpClient

* README fixed with

* Removed NewTestProvider/NewTestInvalidProvider/NewTestNonExistProvider and replace with New; Minor tweaks for Empty URI and Non-exist cases; Added one more test for shutdowned server

* Simplified unit tests for TestNonExistent and TestRetrieveFromShutdownServer

* Update CHANGELOG.md

Co-authored-by: Bogdan Drutu <[email protected]>
Co-authored-by: Bogdan Drutu <[email protected]>

CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### 💡 Enhancements 💡
 
+- Add httpprovider to allow loading config files stored in HTTP (#5810)
+
 ### 🧰 Bug fixes 🧰
 
 - Fix bug in setting the correct collector state after a configuration change event. (#5830)

confmap/provider/httpprovider/README.md

@@ -0,0 +1,13 @@
+What is this new component httpprovider?
+- An implementation of `confmap.Provider` for HTTP (httpprovider) allows OTEL Collector the ability to load configuration for itself by fetching and reading config files stored in HTTP servers.
+
+How this new component httpprovider works?
+- It will be called by `confmap.Resolver` to load configurations for OTEL Collector.
+- By giving a config URI starting with prefix 'http://', this httpprovider will be used to download config files from given HTTP URIs, and then used the downloaded config files to deploy the OTEL Collector.
+- In our code, we check the validity scheme and string pattern of HTTP URIs. And also check if there are any problems on config downloading and config deserialization.
+
+Expected URI format:
+- http://...
+
+Prerequistes:
+- Need to setup a HTTP server ahead, which returns with a config files according to the given URI

confmap/provider/httpprovider/provider.go

@@ -0,0 +1,80 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpprovider // import "go.opentelemetry.io/collector/confmap/provider/httpprovider"
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/confmap/provider/internal"
+)
+
+const (
+	schemeName = "http"
+)
+
+type provider struct {
+	client http.Client
+}
+
+// New returns a new confmap.Provider that reads the configuration from a file.
+//
+// This Provider supports "http" scheme, and can be called with a "uri" that follows:
+//
+// One example for http-uri be like: http://localhost:3333/getConfig
+//
+// Examples:
+// `http://localhost:3333/getConfig` - (unix, windows)
+func New() confmap.Provider {
+	return &provider{client: http.Client{}}
+}
+
+func (fmp *provider) Retrieve(_ context.Context, uri string, _ confmap.WatcherFunc) (*confmap.Retrieved, error) {
+	if !strings.HasPrefix(uri, schemeName+":") {
+		return nil, fmt.Errorf("%q uri is not supported by %q provider", uri, schemeName)
+	}
+
+	// send a HTTP GET request
+	resp, err := fmp.client.Get(uri)
+	if err != nil {
+		return nil, fmt.Errorf("unable to download the file via HTTP GET for uri %q, with err: %w ", uri, err)
+	}
+	defer resp.Body.Close()
+
+	// check the HTTP status code
+	if resp.StatusCode != 200 {
+		return nil, fmt.Errorf("404: resource didn't exist, fail to read the response body from uri %q", uri)
+	}
+
+	// read the response body
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("fail to read the response body from uri %q, with err: %w ", uri, err)
+	}
+
+	return internal.NewRetrievedFromYAML(body)
+}
+
+func (*provider) Scheme() string {
+	return schemeName
+}
+
+func (*provider) Shutdown(context.Context) error {
+	return nil
+}

confmap/provider/httpprovider/provider_test.go

@@ -0,0 +1,116 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpprovider
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+)
+
+func TestFunctionalityDownloadFileHTTP(t *testing.T) {
+	fp := New()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		f, err := os.ReadFile("./testdata/otel-config.yaml")
+		if err != nil {
+			w.WriteHeader(404)
+			_, innerErr := w.Write([]byte("Cannot find the config file"))
+			if innerErr != nil {
+				fmt.Println("Write failed: ", innerErr)
+			}
+			return
+		}
+		w.WriteHeader(200)
+		_, err = w.Write(f)
+		if err != nil {
+			fmt.Println("Write failed: ", err)
+		}
+	}))
+	defer ts.Close()
+	_, err := fp.Retrieve(context.Background(), ts.URL, nil)
+	assert.NoError(t, err)
+	assert.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestUnsupportedScheme(t *testing.T) {
+	fp := New()
+	_, err := fp.Retrieve(context.Background(), "https://...", nil)
+	assert.Error(t, err)
+	assert.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestEmptyURI(t *testing.T) {
+	fp := New()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(400)
+	}))
+	defer ts.Close()
+	_, err := fp.Retrieve(context.Background(), ts.URL, nil)
+	require.Error(t, err)
+	require.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestRetrieveFromShutdownServer(t *testing.T) {
+	fp := New()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	ts.Close()
+	_, err := fp.Retrieve(context.Background(), ts.URL, nil)
+	assert.Error(t, err)
+	require.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestNonExistent(t *testing.T) {
+	fp := New()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(404)
+	}))
+	defer ts.Close()
+	_, err := fp.Retrieve(context.Background(), ts.URL, nil)
+	assert.Error(t, err)
+	require.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestInvalidYAML(t *testing.T) {
+	fp := New()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+		_, err := w.Write([]byte("wrong : ["))
+		if err != nil {
+			fmt.Println("Write failed: ", err)
+		}
+	}))
+	defer ts.Close()
+	_, err := fp.Retrieve(context.Background(), ts.URL, nil)
+	assert.Error(t, err)
+	require.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestScheme(t *testing.T) {
+	fp := New()
+	assert.Equal(t, "http", fp.Scheme())
+	require.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestValidateProviderScheme(t *testing.T) {
+	assert.NoError(t, confmaptest.ValidateProviderScheme(New()))
+}

confmap/provider/httpprovider/testdata/otel-config.yaml

@@ -0,0 +1,37 @@
+extensions:
+  memory_ballast:
+    size_mib: 512
+  zpages:
+    endpoint: 0.0.0.0:55679
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+
+processors:
+  batch:
+  memory_limiter:
+    # 75% of maximum memory up to 4G
+    limit_mib: 1536
+    # 25% of limit up to 2G
+    spike_limit_mib: 512
+    check_interval: 5s
+
+exporters:
+  logging:
+    loglevel: debug
+
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [memory_limiter, batch]
+      exporters: [logging]
+    metrics:
+      receivers: [otlp]
+      processors: [memory_limiter, batch]
+      exporters: [logging]
+
+  extensions: [memory_ballast, zpages]

service/config_provider.go

@@ -23,6 +23,7 @@ import (
 	"go.opentelemetry.io/collector/confmap/converter/expandconverter"
 	"go.opentelemetry.io/collector/confmap/provider/envprovider"
 	"go.opentelemetry.io/collector/confmap/provider/fileprovider"
+	"go.opentelemetry.io/collector/confmap/provider/httpprovider"
 	"go.opentelemetry.io/collector/confmap/provider/yamlprovider"
 	"go.opentelemetry.io/collector/service/internal/configunmarshaler"
 )
@@ -84,7 +85,7 @@ func newDefaultConfigProviderSettings(uris []string) ConfigProviderSettings {
 	return ConfigProviderSettings{
 		ResolverSettings: confmap.ResolverSettings{
 			URIs:       uris,
-			Providers:  makeMapProvidersMap(fileprovider.New(), envprovider.New(), yamlprovider.New()),
+			Providers:  makeMapProvidersMap(fileprovider.New(), envprovider.New(), yamlprovider.New(), httpprovider.New()),
 			Converters: []confmap.Converter{expandconverter.New()},
 		},
 	}