split out manager provider

Author: ankitpatel96

Diff for: processor/schemaprocessor/internal/translation/manager.go

@@ -0,0 +1,131 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package translation // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/translation"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"go.uber.org/zap"
+)
+
+var errNilValueProvided = errors.New("nil value provided")
+
+// Manager is responsible for ensuring that schemas are kept up to date
+// with the most recent version that are requested.
+type Manager interface {
+	// RequestTranslation will provide either the defined Translation
+	// if it is a known target, or, return a noop variation.
+	// In the event that a matched Translation, on a missed version
+	// there is a potential to block during this process.
+	// Otherwise, the translation will allow concurrent reads.
+	RequestTranslation(ctx context.Context, schemaURL string) Translation
+
+	// SetProviders will update the list of providers used by the manager
+	// to look up schemaURLs
+	SetProviders(providers ...Provider) error
+}
+
+type manager struct {
+	log *zap.Logger
+
+	rw           sync.RWMutex
+	providers    []Provider
+	match        map[string]*Version
+	translations map[string]*translator
+}
+
+var _ Manager = (*manager)(nil)
+
+// NewManager creates a manager that will allow for management
+// of schema, the options allow for additional properties to be
+// added to manager to enable additional locations of where to check
+// for translations file.
+func NewManager(targets []string, log *zap.Logger) (Manager, error) {
+	if log == nil {
+		return nil, fmt.Errorf("logger: %w", errNilValueProvided)
+	}
+
+	match := make(map[string]*Version, len(targets))
+	for _, target := range targets {
+		family, version, err := GetFamilyAndVersion(target)
+		if err != nil {
+			return nil, err
+		}
+		match[family] = version
+	}
+
+	return &manager{
+		log:          log,
+		match:        match,
+		translations: make(map[string]*translator),
+	}, nil
+}
+
+func (m *manager) RequestTranslation(ctx context.Context, schemaURL string) Translation {
+	family, version, err := GetFamilyAndVersion(schemaURL)
+	if err != nil {
+		m.log.Debug("No valid schema url was provided, using no-op schema",
+			zap.String("schema-url", schemaURL),
+		)
+		return nopTranslation{}
+	}
+
+	target, match := m.match[family]
+	if !match {
+		m.log.Debug("Not a known target, providing Nop Translation",
+			zap.String("schema-url", schemaURL),
+		)
+		return nopTranslation{}
+	}
+
+	m.rw.RLock()
+	t, exists := m.translations[family]
+	m.rw.RUnlock()
+
+	if exists && t.SupportedVersion(version) {
+		return t
+	}
+
+	for _, p := range m.providers {
+		content, err := p.Lookup(ctx, schemaURL)
+		if err != nil {
+			m.log.Error("Failed to lookup schemaURL",
+				zap.Error(err),
+				zap.String("schemaURL", schemaURL),
+			)
+			// todo(ankit) figure out what to do when the providers dont respond something good
+		}
+		t, err := newTranslatorFromReader(
+			m.log.Named("translator").With(
+				zap.String("family", family),
+				zap.Stringer("target", target),
+			),
+			joinSchemaFamilyAndVersion(family, target),
+			content,
+		)
+		if err != nil {
+			m.log.Error("Failed to create translator", zap.Error(err))
+			continue
+		}
+		m.rw.Lock()
+		m.translations[family] = t
+		m.rw.Unlock()
+		return t
+	}
+
+	return nopTranslation{}
+}
+
+func (m *manager) SetProviders(providers ...Provider) error {
+	if len(providers) == 0 {
+		return fmt.Errorf("zero providers set: %w", errNilValueProvided)
+	}
+	m.rw.Lock()
+	m.providers = append(m.providers[:0], providers...)
+	m.rw.Unlock()
+	return nil
+}

Diff for: processor/schemaprocessor/internal/translation/manager_test.go

@@ -0,0 +1,68 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package translation
+
+import (
+	"context"
+	_ "embed"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
+)
+
+//go:embed testdata/schema.yaml
+var exampleTranslation []byte
+
+func TranslationHandler(t *testing.T) http.Handler {
+	assert.NotEmpty(t, exampleTranslation, "SchemaContent MUST not be empty")
+	return http.HandlerFunc(func(wr http.ResponseWriter, _ *http.Request) {
+		_, err := wr.Write(exampleTranslation)
+		assert.NoError(t, err, "Must not have issues writing schema content")
+	})
+}
+
+func TestRequestTranslation(t *testing.T) {
+	t.Parallel()
+
+	s := httptest.NewServer(TranslationHandler(t))
+	t.Cleanup(s.Close)
+
+	schemaURL := fmt.Sprintf("%s/1.1.0", s.URL)
+
+	m, err := NewManager(
+		[]string{schemaURL},
+		zaptest.NewLogger(t),
+	)
+	require.NoError(t, err, "Must not error when created manager")
+	require.NoError(t, m.SetProviders(NewHTTPProvider(s.Client())), "Must have no issues trying to set providers")
+
+	nop, ok := m.RequestTranslation(context.Background(), "/not/a/valid/schema/URL").(nopTranslation)
+	require.True(t, ok, "Must return a NoopTranslation if no valid schema URL is provided")
+	require.NotNil(t, nop, "Must have a valid translation")
+
+	tn, ok := m.RequestTranslation(context.Background(), schemaURL).(*translator)
+	require.True(t, ok, "Can cast to the concrete type")
+	require.NotNil(t, tn, "Must have a valid translation")
+
+	assert.True(t, tn.SupportedVersion(&Version{1, 0, 0}), "Must have the version listed as supported")
+
+	count := 0
+	prevRev := &Version{1, 0, 0}
+	it, status := tn.iterator(context.Background(), prevRev)
+	assert.Equal(t, Update, status, "Must return a status of update")
+	for currRev, more := it(); more; currRev, more = it() {
+		assert.True(t, prevRev.LessThan(currRev.Version()))
+		prevRev = currRev.Version()
+		count++
+	}
+
+	tn, ok = m.RequestTranslation(context.Background(), schemaURL).(*translator)
+	require.True(t, ok, "Can cast to the concrete type")
+	require.NotNil(t, tn, "Must have a valid translation")
+}

Diff for: processor/schemaprocessor/internal/translation/provider.go

@@ -0,0 +1,73 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package translation // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/translation"
+
+import (
+	"bytes"
+	"context"
+	"embed"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+)
+
+// Provider allows for collector extensions to be used to look up schemaURLs
+type Provider interface {
+	// Lookup whill check the underlying provider to see if content exists
+	// for the provided schemaURL, in the even that it doesn't an error is returned.
+	Lookup(ctx context.Context, schemaURL string) (content io.Reader, err error)
+}
+
+type httpProvider struct {
+	client *http.Client
+}
+
+var _ Provider = (*httpProvider)(nil)
+
+func NewHTTPProvider(client *http.Client) Provider {
+	return &httpProvider{client: client}
+}
+
+func (hp *httpProvider) Lookup(ctx context.Context, schemaURL string) (io.Reader, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, schemaURL, http.NoBody)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := hp.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	content := bytes.NewBuffer(nil)
+	if _, err := content.ReadFrom(resp.Body); err != nil {
+		return nil, err
+	}
+	if err := resp.Body.Close(); err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("invalid status code returned: %d", resp.StatusCode)
+	}
+	return content, nil
+}
+
+type testProvider struct {
+	fs *embed.FS
+}
+
+func NewTestProvider(fs *embed.FS) Provider {
+	return &testProvider{fs: fs}
+}
+
+func (tp testProvider) Lookup(_ context.Context, schemaURL string) (io.Reader, error) {
+	parsedPath, err := url.Parse(schemaURL)
+	if err != nil {
+		return nil, err
+	}
+	f, err := tp.fs.Open(parsedPath.Path[1:])
+	if err != nil {
+		return nil, err
+	}
+	return f, nil
+}

Diff for: processor/schemaprocessor/internal/translation/provider_test.go

@@ -0,0 +1,52 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package translation
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestInvalidHTTPProviderTests(t *testing.T) {
+	t.Parallel()
+
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.RequestURI != "/1.7.0" {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		_, err := io.Copy(w, LoadTranslationVersion(t, "complex_changeset.yml"))
+		assert.NoError(t, err, "Must not error when trying load dataset")
+	}))
+	t.Cleanup(s.Close)
+
+	tests := []struct {
+		scenario string
+		url      string
+	}{
+		{
+			scenario: "A failed request happens",
+			url:      fmt.Sprint(s.URL, "/not/a/valid/path/1.7.0"),
+		},
+		{
+			scenario: "invalid url",
+			url:      "unix:///localhost",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.scenario, func(t *testing.T) {
+			p := NewHTTPProvider(s.Client())
+			content, err := p.Lookup(context.Background(), tc.url)
+			assert.Nil(t, content, "Expected to be nil")
+			assert.Error(t, err, "Must have errored processing request")
+		})
+	}
+}