fix(processor/schema): fetch target schema URL on upgrade path (#47632)

#### Description

When upgrading a signal (incoming version older than the configured
target), the manager was fetching the incoming signal's schema URL to
retrieve the schema file. Real OTel schema files only contain history up
to their own version — an incoming `1.21.0` schema file has no knowledge
of changes made in `1.22.0`–`1.27.0`. The translator found no matching
revisions and silently returned `NoChange`, meaning no attributes were
ever renamed on the upgrade path.

The fix: for upgrades, fetch the **target** schema URL instead. The
target schema file contains the complete forward migration history.
Downgrades correctly continue to fetch the incoming signal's (newer)
URL, since that file contains the history needed to reverse changes.

#### Link to tracking issue

Fixes #47427

#### Testing

Added `TestRequestTranslationUpgrade` in `manager_test.go`. It uses a
URL-aware HTTP server that serves a stub schema (no forward history) for
old version URLs and the full schema only at the target URL. This
proves:

1. The fix correctly fetches the target URL on the upgrade path
2. The attribute rename defined in the target schema
(`db.cassandra.keyspace` → `db.name`) is applied
3. Without the fix, the test fails because the stub schema has no
revisions to apply

#### Documentation

No documentation changes required — this is a bug fix for behaviour that
was never working correctly.

Author: MikeGoldsmith

.chloggen/schema-processor-upgrade-url.yaml

@@ -0,0 +1,11 @@
+change_type: bug_fix
+component: processor/schema
+note: Fix upgrade path silently applying no changes due to fetching the incoming signal's schema file instead of the target schema file
+issues: [47427, 47632]
+subtext: |
+  When upgrading a signal to a newer target version, the processor was fetching the schema
+  file at the incoming signal's URL. Because OTel schema files only contain history up to
+  their own version, the translator found no applicable revisions and silently passed the
+  signal through unchanged. The processor now fetches the target schema URL on the upgrade
+  path, which contains the complete forward migration history.
+change_logs: [user]

processor/schemaprocessor/internal/translation/manager.go

@@ -95,12 +95,20 @@ func (m *manager) RequestTranslation(ctx context.Context, schemaURL string) (Tra
 		return t, nil
 	}
 
+	// Always fetch the schema file for the higher version, since it contains the complete
+	// migration history. For upgrades (signal older than target), fetch the target URL.
+	// For downgrades (signal newer than target), fetch the signal URL.
+	fetchURL := schemaURL
+	if version.Compare(targetTranslation) == Update {
+		fetchURL = joinSchemaFamilyAndVersion(family, targetTranslation)
+	}
+
 	for _, p := range m.providers {
-		content, err := p.Retrieve(ctx, schemaURL)
+		content, err := p.Retrieve(ctx, fetchURL)
 		if err != nil {
 			m.log.Error("Failed to lookup schemaURL",
 				zap.Error(err),
-				zap.String("schemaURL", schemaURL),
+				zap.String("schemaURL", fetchURL),
 			)
 			// If we fail to retrieve the schema, we should
 			// try the next provider

processor/schemaprocessor/internal/translation/manager_test.go

@@ -10,10 +10,12 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -66,6 +68,70 @@ func TestRequestTranslation(t *testing.T) {
 	}
 }
 
+// TestRequestTranslationUpgrade verifies that the upgrade path fetches the target schema
+// URL (not the incoming signal's URL). Real OTel schema files only contain history up to
+// their own version, so an incoming 1.0.0 schema file has no knowledge of changes made
+// in 1.1.0–1.9.0. The processor must fetch the target schema file to get the full history.
+func TestRequestTranslationUpgrade(t *testing.T) {
+	t.Parallel()
+
+	// fullSchema contains history for 1.0.0–1.9.0, including a rename in 1.8.0.
+	// stubSchema simulates what a real 1.0.0 schema file looks like: it only knows
+	// about itself and has no forward history.
+	fullSchema := LoadTranslationVersion(t, TranslationVersion190)
+	stubSchema := `file_format: 1.0.0
+schema_url: https://example.com/1.0.0
+versions:
+  1.0.0:
+`
+
+	// URL-aware server: serve the full schema only at the target URL.
+	// All other URLs (simulating older signal schema files) return the stub.
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var content string
+		if strings.HasSuffix(r.URL.Path, TranslationVersion190) {
+			content = fullSchema
+		} else {
+			content = stubSchema
+		}
+		_, err := w.Write([]byte(content))
+		assert.NoError(t, err)
+	}))
+	t.Cleanup(s.Close)
+
+	targetURL := fmt.Sprintf("%s/%s", s.URL, TranslationVersion190)
+	signalURL := fmt.Sprintf("%s/1.0.0", s.URL)
+
+	m, err := NewManager(
+		[]string{targetURL},
+		zaptest.NewLogger(t),
+		NewHTTPProvider(s.Client()),
+	)
+	require.NoError(t, err)
+
+	tr, err := m.RequestTranslation(t.Context(), signalURL)
+	require.NoError(t, err, "Must not error on upgrade path")
+	require.NotNil(t, tr)
+
+	// The translator must know about 1.0.0 (loaded from the target schema file).
+	assert.True(t, tr.SupportedVersion(&Version{1, 0, 0}), "Must support the incoming signal version")
+
+	// Apply the upgrade: 1.8.0 in the target schema renames db.cassandra.keyspace → db.name.
+	scopeSpans := ptrace.NewScopeSpans()
+	scopeSpans.SetSchemaUrl(signalURL)
+	span := scopeSpans.Spans().AppendEmpty()
+	span.Attributes().PutStr("db.cassandra.keyspace", "my_keyspace")
+
+	require.NoError(t, tr.ApplyScopeSpanChanges(scopeSpans, signalURL))
+
+	val, ok := span.Attributes().Get("db.name")
+	require.True(t, ok, "db.name must exist after upgrade — rename was not applied")
+	assert.Equal(t, "my_keyspace", val.Str())
+
+	_, oldExists := span.Attributes().Get("db.cassandra.keyspace")
+	assert.False(t, oldExists, "old attribute must be removed after upgrade")
+}
+
 type errorProvider struct{}
 
 func (*errorProvider) Retrieve(_ context.Context, _ string) (string, error) {