[pull] master from netdata:master

.agents/sow/specs/README.md

@@ -16,6 +16,7 @@ research cannot be mistaken for product specification.
 | Spec | Scope | Main consumers |
 |---|---|---|
 | [go-v2-host-scope.md](go-v2-host-scope.md) | Go framework V2 host-scope and vnode metric routing contract. | Go V2 collector work, `metrix`, `jobruntime`, `chartengine`, go.d collector docs and skills. |
+| [journal-log-writer-directory-contract.md](journal-log-writer-directory-contract.md) | `Log::new` requires an absolute journal directory (rejected early otherwise); consumers must resolve relative config against a stable base. | journal-log-writer consumers: otel-plugin logs, netflow-plugin tiers. |
 | [netflow-tier-commit-workers.md](netflow-tier-commit-workers.md) | NetFlow rollup-tier commit worker contract: ownership, doorbell protocol, shutdown order, stretch semantics, lock discipline, telemetry. | netflow-plugin ingest/tiering work, tier benchmark and soak/crash tests, network-flows operator docs. |
 | [prometheus-profiles.md](prometheus-profiles.md) | go.d prometheus chart-profile format, selection modes, app-segment resolution, and chart-context assembly. | prometheus collector profile work, chart-context/UI app-section contract, profile authoring. |
 | [query-planner-tier-selection.md](query-planner-tier-selection.md) | Automatic storage-tier selection for metric query planning. | Query planner code, API data paths, MCP metric query behavior. |

.agents/sow/specs/journal-log-writer-directory-contract.md

@@ -0,0 +1,46 @@
+# journal-log-writer directory contract
+
+## Scope
+
+Durable contract for the `journal-log-writer` crate (`Log::new`) regarding the
+journal directory it is given. Applies to every consumer that constructs a
+journal `Log`: the OTEL logs plugin (`otel-plugin`) and the NetFlow plugin
+(`netflow-plugin`).
+
+## Contract
+
+- **The journal directory passed to `Log::new` MUST be an absolute path.**
+  A non-absolute path is rejected at construction with
+  `WriterError::InvalidPath`, naming the offending value. The rejection happens
+  **before** any filesystem mutation (no directory is created).
+- Rationale: journal file paths are parsed by
+  `journal_registry::repository::File::from_path`, which only accepts absolute
+  paths. A relative directory therefore could never produce a writable journal —
+  previously it failed late, at first write, with a cause-less
+  `failed to create journal file in <path>` error. The early rejection makes the
+  misconfiguration diagnosable at startup.
+
+## Consumer responsibilities
+
+- **Resolve relative configuration before calling `Log::new`.** A consumer that
+  accepts a possibly-relative directory from its own config SHOULD resolve it to
+  an absolute path first, against a stable base directory it controls — never
+  rely on the process working directory. The `Log::new` check is the safety net,
+  not the primary resolution mechanism.
+  - NetFlow resolves `journal_dir` against the netdata cache directory **when one
+    is available**, via `resolve_relative_path`
+    (`netflow-plugin/src/plugin_config/runtime.rs`). Its default is relative
+    (`flows`), so under netdata (where `NETDATA_CACHE_DIR` is set) it resolves to
+    an absolute path. Standalone with no cache dir, the value stays relative and
+    is rejected by the `Log::new` check at startup (fail-fast, not a regression —
+    that configuration never produced a writable journal).
+  - The OTEL logs plugin does not resolve `journal_dir`; it requires an absolute
+    value. Its stock config renders one (`<logdir>/otel/v1` via `@logdir_POST@`);
+    a relative override (`NETDATA_OTEL_LOGS_JOURNAL_DIR` env or a hand-templated
+    config) is treated as operator error and rejected at startup.
+
+## Notes
+
+- `Log::new` still calls `canonicalize()` on the (now guaranteed absolute) path
+  to verify accessibility; its result is intentionally not used to rewrite the
+  stored path (no symlink normalization), to keep behavior minimal.

src/crates/journal-log-writer/src/log/chain.rs

@@ -151,7 +151,8 @@ impl OwnedChain {
         let Some(file) = create_chain_file(&self.path, seqnum_id, head_seqnum, head_realtime)
         else {
             return Err(WriterError::FileCreation(format!(
-                "failed to create journal file in {}",
+                "failed to create journal file in {}: the generated file path \
+                 could not be parsed as a journal repository file",
                 self.path.display()
             )));
         };

src/crates/journal-log-writer/src/log/mod.rs

@@ -23,6 +23,18 @@ const STACK_ENTRY_REF_LIMIT: usize = 128;
 const SOURCE_REALTIME_PREFIX: &[u8] = b"_SOURCE_REALTIME_TIMESTAMP=";
 
 fn create_chain(path: &Path) -> Result<OwnedChain> {
+    // The journal directory must be absolute: downstream file-path parsing
+    // (`repository::File::from_path`) only accepts absolute paths, so a relative
+    // value would otherwise fail later at first write with a cause-less error.
+    // Checked first, before any I/O, so the rejection is FS-free and the error
+    // names the offending value regardless of machine-id availability.
+    if !path.is_absolute() {
+        return Err(WriterError::InvalidPath(format!(
+            "journal directory must be an absolute path, got: {}",
+            path.display()
+        )));
+    }
+
     let machine_id = load_machine_id()
         .map_err(|e| WriterError::MachineId(format!("failed to load machine ID: {}", e)))?;
 

src/crates/journal-log-writer/tests/log_writer.rs

@@ -767,3 +767,67 @@ fn test_lifecycle_observer_reports_missing_retention_deletions() {
         "files removed from chain/accounting must still be reported for retention follow-up"
     );
 }
+
+#[test]
+fn relative_journal_dir_is_rejected_before_any_io() {
+    // A relative journal directory must fail fast at construction with a clear,
+    // actionable error that names the offending value, and must create no
+    // directory (validation happens before any filesystem mutation).
+    let root = "relative-journal-dir-rejected";
+    let dir = format!("{root}/otel/v1");
+
+    // Clean any leftover from a prior run. A real removal error (e.g. permission
+    // denied) must surface rather than mask a spurious assertion below; only a
+    // missing directory is the expected, ignorable case.
+    match fs::remove_dir_all(root) {
+        Ok(()) => {}
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+        Err(e) => panic!("failed to clean up leftover {root}: {e}"),
+    }
+
+    match Log::new(Path::new(&dir), test_config()) {
+        Ok(_) => panic!("expected Log::new to reject a relative journal directory"),
+        Err(e) => {
+            let msg = e.to_string();
+            assert!(
+                msg.contains("absolute"),
+                "error should mention the absolute-path requirement, got: {msg}"
+            );
+            assert!(
+                msg.contains(&dir),
+                "error should name the offending value, got: {msg}"
+            );
+        }
+    }
+
+    assert!(
+        !Path::new(root).exists(),
+        "rejection must happen before any I/O; no directory may be created"
+    );
+}
+
+#[test]
+fn absolute_journal_dir_is_accepted() {
+    let tmp = TempDir::new().expect("create temp dir");
+    assert!(tmp.path().is_absolute());
+    match Log::new(tmp.path(), test_config()) {
+        Ok(_) => {}
+        Err(e) => panic!("absolute journal directory must be accepted, got: {e}"),
+    }
+}
+
+#[test]
+fn nonexistent_absolute_journal_dir_is_created_and_accepted() {
+    // Common production case: the configured absolute directory does not yet
+    // exist on first run and must be created.
+    let tmp = TempDir::new().expect("create temp dir");
+    let fresh = tmp.path().join("not-yet-created/otel/v1");
+    assert!(!fresh.exists());
+
+    match Log::new(&fresh, test_config()) {
+        Ok(_) => {}
+        Err(e) => panic!("non-existent absolute journal directory must be accepted, got: {e}"),
+    }
+
+    assert!(fresh.is_dir(), "journal directory should have been created");
+}

src/crates/netdata-otel/otel-plugin/configs/otel.yaml.in

@@ -35,6 +35,7 @@ metrics:
 
 logs:
   # Directory to store journal files for logs.
+  # Must be an absolute path; a relative value is rejected at startup.
   journal_dir: @logdir_POST@/otel/v1
 
   # Maximum file size for individual journal files (e.g., "100MB", "1.5GB").

src/go/plugin/go.d/collector/snmp_topology/collector.go

@@ -14,7 +14,6 @@ import (
 	"github.com/gosnmp/gosnmp"
 
 	"github.com/netdata/netdata/go/plugins/logger"
-	"github.com/netdata/netdata/go/plugins/pkg/funcapi"
 	"github.com/netdata/netdata/go/plugins/pkg/metrix"
 	"github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
 	"github.com/netdata/netdata/go/plugins/plugin/go.d/collector/snmp/ddsnmp"
@@ -81,7 +80,6 @@ type (
 
 		deviceCaches        map[string]*topologyCache // one cache per SNMP device
 		deviceLastCollected map[string]time.Time      // last collection time per device
-		topologyCache       *topologyCache            // current device cache (set during refreshDeviceTopology)
 		topologyRegistry    *topologyRegistry
 		deviceSource        deviceSource
 		trapEnrichment      *TrapEnrichmentHandle
@@ -322,17 +320,15 @@ func (c *Collector) refreshDeviceTopology(ctx context.Context, key string, dev d
 	// Build the next snapshot off-registry. Function readers keep seeing the
 	// previous complete snapshot until this collection is fully ingested.
 	next := c.newDeviceCollectionCache(dev)
-	c.topologyCache = next
-	defer func() { c.topologyCache = nil }()
 
-	c.updateTopologySysUptime(sysUptime)
-	c.updateTopologyProfileTags(pms)
-	c.ingestTopologyProfileMetrics(pms)
-	c.collectTopologyVTPVLANContexts(ctx, dev)
+	next.updateTopologySysUptime(sysUptime)
+	next.updateTopologyProfileTags(pms)
+	next.ingestTopologyProfileMetrics(pms)
+	c.collectTopologyVTPVLANContexts(ctx, next, dev)
 	if ctx.Err() != nil {
 		return false
 	}
-	c.finalizeTopologyCache()
+	c.finalizeTopologyCache(next)
 
 	cache := c.getOrCreateDeviceCache(key)
 	cache.mu.Lock()
@@ -398,18 +394,6 @@ func (c *Collector) getTopologyProfiles(dev ddsnmp.DeviceConnectionInfo) []*ddsn
 	return c.findTopologyProfiles(dev)
 }
 
-func (c *Collector) ingestTopologyProfileMetrics(pms []*ddsnmp.ProfileMetrics) {
-	for _, pm := range pms {
-		c.ingestTopologyMetricSet(pm.TopologyMetrics)
-	}
-}
-
-func (c *Collector) ingestTopologyMetricSet(metrics []ddsnmp.Metric) {
-	for _, metric := range metrics {
-		c.updateTopologyCacheEntry(metric)
-	}
-}
-
 func newSNMPClientFromDeviceInfo(newClient func() gosnmp.Handler, dev ddsnmp.DeviceConnectionInfo) (gosnmp.Handler, error) {
 	client := newClient()
 
@@ -450,20 +434,3 @@ func newSNMPClientFromDeviceInfo(newClient func() gosnmp.Handler, dev ddsnmp.Dev
 
 	return client, nil
 }
-
-func topologyMethods() []funcapi.MethodConfig {
-	return []funcapi.MethodConfig{
-		topologyMethodConfig(),
-	}
-}
-
-func topologyFunctionHandler(job collectorapi.RuntimeJob) funcapi.MethodHandler {
-	if job == nil {
-		return nil
-	}
-	coll, ok := job.Collector().(*Collector)
-	if !ok || coll == nil {
-		return nil
-	}
-	return &funcTopology{registry: coll.topologyRegistry}
-}

src/go/plugin/go.d/collector/snmp_topology/collector_refresh_test.go

@@ -4,6 +4,7 @@ package snmptopology
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"testing"
 	"time"
@@ -359,6 +360,42 @@ func TestCollector_RefreshKeepsPublishedSnapshotWhileCollectionRuns(t *testing.T
 	<-done
 }
 
+func TestCollector_RefreshFailureKeepsPublishedSnapshot(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	dev := ddsnmp.DeviceConnectionInfo{
+		Hostname:    "10.0.0.10",
+		Port:        161,
+		SysObjectID: "1.3.6.1.4.1.9.1.1",
+	}
+	mockHandler := snmpmock.NewMockHandler(ctrl)
+	expectTopologyRefreshSNMPClientConnect(mockHandler, dev)
+	mockHandler.EXPECT().Close().Return(nil)
+
+	key := "10.0.0.10:161"
+	published := newTopologyCache()
+	seedPublishedEndpointSnapshot(published)
+
+	coll := newTestSNMPTopologyCollector()
+	coll.deviceCaches[key] = published
+	coll.topologyRegistry.register(published)
+	coll.newSnmpClient = func() gosnmp.Handler { return mockHandler }
+	coll.newDdSnmpColl = func(ddsnmpcollector.Config) ddCollector {
+		return ddCollectorFunc(func() ([]*ddsnmp.ProfileMetrics, error) {
+			return nil, errors.New("collection failed")
+		})
+	}
+
+	require.False(t, coll.refreshDeviceTopology(context.Background(), key, dev))
+
+	snapshot, ok := published.snapshotEngineObservations()
+	require.True(t, ok)
+	require.Len(t, snapshot.l2Observations, 1)
+	require.Len(t, snapshot.l2Observations[0].FDBEntries, 1)
+	require.Len(t, snapshot.l2Observations[0].ARPNDEntries, 1)
+}
+
 type blockingTopologyCollector struct {
 	started chan<- struct{}
 	release <-chan struct{}
@@ -376,15 +413,7 @@ func (c *blockingTopologyCollector) Collect() ([]*ddsnmp.ProfileMetrics, error)
 }
 
 func expectTopologyRefreshSNMPClient(mockHandler *snmpmock.MockHandler, dev ddsnmp.DeviceConnectionInfo) {
-	mockHandler.EXPECT().SetTarget(dev.Hostname)
-	mockHandler.EXPECT().SetPort(uint16(dev.Port))
-	mockHandler.EXPECT().SetRetries(dev.Retries)
-	mockHandler.EXPECT().SetTimeout(time.Duration(dev.Timeout) * time.Second)
-	mockHandler.EXPECT().SetMaxOids(dev.MaxOIDs)
-	mockHandler.EXPECT().SetMaxRepetitions(uint32(dev.MaxRepetitions))
-	mockHandler.EXPECT().SetCommunity(dev.Community)
-	mockHandler.EXPECT().SetVersion(gosnmp.Version2c)
-	mockHandler.EXPECT().Connect().Return(nil)
+	expectTopologyRefreshSNMPClientConnect(mockHandler, dev)
 	mockHandler.EXPECT().Get(gomock.InAnyOrder([]string{
 		snmputils.OidSnmpEngineTime,
 		snmputils.OidHrSystemUptime,
@@ -397,6 +426,18 @@ func expectTopologyRefreshSNMPClient(mockHandler *snmpmock.MockHandler, dev ddsn
 	mockHandler.EXPECT().Close().Return(nil)
 }
 
+func expectTopologyRefreshSNMPClientConnect(mockHandler *snmpmock.MockHandler, dev ddsnmp.DeviceConnectionInfo) {
+	mockHandler.EXPECT().SetTarget(dev.Hostname)
+	mockHandler.EXPECT().SetPort(uint16(dev.Port))
+	mockHandler.EXPECT().SetRetries(dev.Retries)
+	mockHandler.EXPECT().SetTimeout(time.Duration(dev.Timeout) * time.Second)
+	mockHandler.EXPECT().SetMaxOids(dev.MaxOIDs)
+	mockHandler.EXPECT().SetMaxRepetitions(uint32(dev.MaxRepetitions))
+	mockHandler.EXPECT().SetCommunity(dev.Community)
+	mockHandler.EXPECT().SetVersion(gosnmp.Version2c)
+	mockHandler.EXPECT().Connect().Return(nil)
+}
+
 func seedPublishedEndpointSnapshot(cache *topologyCache) {
 	now := time.Now()
 	cache.updateTime = now