[synthetics][SYNTH-24584]: Support On-demand Network Tests (#46789)

### What does this PR do?

Adds support for On-demand Network tests

* Creates a new "OnDemandPoller" and do a double queue with Scheduled and On-demand tests
* On-demand tests have higher priority on execution

### Describe how you validated your changes

* Unit Tests
* Local running of the Agent

### Additional Notes


Co-authored-by: rafael.oliveira <rafael.oliveira@datadoghq.com>

Author: rafamelo12

comp/syntheticstestscheduler/common/data.go

@@ -78,6 +78,7 @@ type SyntheticsTestConfig struct {
 	OrgID    int    `json:"org_id"`
 	MainDC   string `json:"main_dc"`
 	PublicID string `json:"public_id"`
+	ResultID string `json:"result_id"`
 	RunType  string `json:"run_type"`
 }
 
@@ -148,6 +149,7 @@ func (c *SyntheticsTestConfig) UnmarshalJSON(data []byte) error {
 		OrgID    int    `json:"org_id"`
 		MainDC   string `json:"main_dc"`
 		PublicID string `json:"public_id"`
+		ResultID string `json:"result_id"`
 		RunType  string `json:"run_type"`
 		Interval int    `json:"tick_every"`
 	}
@@ -162,6 +164,7 @@ func (c *SyntheticsTestConfig) UnmarshalJSON(data []byte) error {
 	c.OrgID = tmp.OrgID
 	c.MainDC = tmp.MainDC
 	c.PublicID = tmp.PublicID
+	c.ResultID = tmp.ResultID
 	c.RunType = tmp.RunType
 	c.Interval = tmp.Interval
 	c.Config.Assertions = tmp.Config.Assertions

comp/syntheticstestscheduler/impl/config.go

@@ -6,11 +6,15 @@
 package syntheticstestschedulerimpl
 
 import (
+	"net/http"
 	"time"
 
 	"github.com/DataDog/datadog-agent/comp/core/config"
+	httputils "github.com/DataDog/datadog-agent/pkg/util/http"
 )
 
+const defaultSite = "datadoghq.com"
+
 type schedulerConfigs struct {
 	workers                    int
 	flushInterval              time.Duration
@@ -24,3 +28,22 @@ func newSchedulerConfigs(agentConfig config.Component) *schedulerConfigs {
 		flushInterval:              agentConfig.GetDuration("synthetics.collector.flush_interval"),
 	}
 }
+
+type onDemandPollerConfig struct {
+	site          string
+	apiKey        string
+	httpTransport *http.Transport
+}
+
+func newOnDemandPollerConfig(agentConfig config.Component) *onDemandPollerConfig {
+	site := agentConfig.GetString("site")
+	if site == "" {
+		site = defaultSite
+	}
+
+	return &onDemandPollerConfig{
+		site:          site,
+		apiKey:        agentConfig.GetString("api_key"),
+		httpTransport: httputils.CreateHTTPTransport(agentConfig),
+	}
+}

comp/syntheticstestscheduler/impl/ondemand.go

@@ -0,0 +1,126 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-present Datadog, Inc.
+
+package syntheticstestschedulerimpl
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/DataDog/datadog-agent/comp/core/hostname"
+	log "github.com/DataDog/datadog-agent/comp/core/log/def"
+	"github.com/DataDog/datadog-agent/comp/syntheticstestscheduler/common"
+)
+
+const (
+	pollingFrequency   = 2 * time.Second
+	httpRequestTimeout = 10 * time.Second
+)
+
+type onDemandTestResponse struct {
+	Tests []common.SyntheticsTestConfig `json:"tests"`
+}
+
+type onDemandPoller struct {
+	httpClient      *http.Client
+	endpoint        string
+	apiKey          string
+	hostNameService hostname.Component
+	log             log.Component
+	timeNowFn       func() time.Time
+	TestsChan       chan SyntheticsTestCtx
+	done            chan struct{}
+}
+
+func newOnDemandPoller(config *onDemandPollerConfig, hostNameService hostname.Component, logger log.Component, timeNowFn func() time.Time) *onDemandPoller {
+	return &onDemandPoller{
+		httpClient:      &http.Client{Transport: config.httpTransport, Timeout: httpRequestTimeout},
+		endpoint:        "https://intake.synthetics." + config.site + "/api/unstable/synthetics/agents/tests",
+		apiKey:          config.apiKey,
+		hostNameService: hostNameService,
+		log:             logger,
+		timeNowFn:       timeNowFn,
+		TestsChan:       make(chan SyntheticsTestCtx, 100),
+		done:            make(chan struct{}),
+	}
+}
+
+func (p *onDemandPoller) start(ctx context.Context) {
+	go p.pollLoop(ctx)
+}
+
+func (p *onDemandPoller) stop() {
+	<-p.done
+}
+
+func (p *onDemandPoller) pollLoop(ctx context.Context) {
+	defer close(p.done)
+
+	ticker := time.NewTicker(pollingFrequency)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			tests, err := p.fetchTests(ctx)
+			if err != nil {
+				p.log.Debugf("error fetching on-demand tests: %s", err)
+				continue
+			}
+			for _, test := range tests {
+				select {
+				case p.TestsChan <- SyntheticsTestCtx{
+					nextRun: p.timeNowFn(),
+					cfg:     test,
+				}:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}
+}
+
+func (p *onDemandPoller) fetchTests(ctx context.Context) ([]common.SyntheticsTestConfig, error) {
+	hostname, err := p.hostNameService.Get(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("error getting hostname: %w", err)
+	}
+
+	rawURL := p.endpoint + "?agent_hostname=" + url.QueryEscape(hostname)
+	u, err := url.Parse(rawURL)
+	if err != nil {
+		return nil, fmt.Errorf("invalid endpoint URL: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("DD-API-KEY", p.apiKey)
+
+	resp, err := p.httpClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	}
+
+	var response onDemandTestResponse
+	if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
+		return nil, err
+	}
+
+	return response.Tests, nil
+}