Skip to content

Commit 7c48cf0

Browse files
jajeffriesjajeffries
committed
makes ports configurable
1 parent 258e7e5 commit 7c48cf0

File tree

5 files changed

+42
-23
lines changed

5 files changed

+42
-23
lines changed

‎agent/agent_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ func TestAgentStart_AutoConfiguresOTLP_WhenFleetActive_CommonBackendExists_OTLPS
112112
Backends: map[string]any{
113113
"common": map[string]any{
114114
"otlp": map[string]any{
115-
"http": "localhost:4318",
115+
"http": "localhost:4319",
116116
},
117117
},
118118
},
@@ -142,7 +142,7 @@ func TestAgentStart_AutoConfiguresOTLP_WhenFleetActive_CommonBackendExists_OTLPS
142142
// Verify that grpc URL was added to existing otlp section
143143
// Note: startBackends deletes "common" from Backends map, but it reads it first into backendsCommon
144144
assert.Equal(t, "localhost:4317", agent.backendsCommon.Otlp.Grpc, "grpc URL should be auto-configured")
145-
assert.Equal(t, "localhost:4318", agent.backendsCommon.Otlp.HTTP, "existing http URL should be preserved")
145+
assert.Equal(t, "localhost:4319", agent.backendsCommon.Otlp.HTTP, "existing http URL should be preserved")
146146
}
147147

148148
func TestAgentStart_AutoConfiguresOTLP_WhenFleetActive_NoCommonBackend(t *testing.T) {

‎agent/config/types.go‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -35,12 +35,14 @@ type GitManager struct {
3535

3636
// FleetManager represents the Orb ConfigManager configuration.
3737
type FleetManager struct {
38-
URL string `yaml:"url"`
39-
TokenURL string `yaml:"token_url"`
40-
Timeout *int `yaml:"timeout,omitempty"`
41-
SkipTLS bool `yaml:"skip_tls"`
42-
ClientID string `yaml:"client_id"`
43-
ClientSecret string `yaml:"client_secret"`
38+
URL string `yaml:"url"`
39+
TokenURL string `yaml:"token_url"`
40+
Timeout *int `yaml:"timeout,omitempty"`
41+
SkipTLS bool `yaml:"skip_tls"`
42+
ClientID string `yaml:"client_id"`
43+
ClientSecret string `yaml:"client_secret"`
44+
OTLPPort *int `yaml:"otlp_port,omitempty"` // Port for main OTLP bridge (default: 4317)
45+
TelemetryPort *int `yaml:"telemetry_port,omitempty"` // Port for telemetry OTLP bridge (default: 4319)
4446
}
4547

4648
// Sources represents the configuration for manager sources, including cloud, local and git.

‎agent/configmgr/fleet.go‎

Lines changed: 27 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"fmt"
66
"log/slog"
7+
"strconv"
78
"time"
89

910
"github.com/eclipse/paho.golang/autopaho"
@@ -138,51 +139,67 @@ func (fleetManager *fleetConfigManager) Start(cfg config.Config, backends map[st
138139
fleetManager.connection.AddOnReadyHook(func(cm *autopaho.ConnectionManager, topics fleet.TokenResponseTopics) {
139140
fleetManager.logger.Info("MQTT connection ready, initializing OTLP bridges")
140141

142+
// Get configurable ports with defaults
143+
otlpPort := getOTLPPort(cfg.OrbAgent.ConfigManager.Sources.Fleet.OTLPPort, 4317)
144+
telemetryPort := getOTLPPort(cfg.OrbAgent.ConfigManager.Sources.Fleet.TelemetryPort, 4319)
145+
141146
// Create publisher adapter (shared by both bridges)
142147
pub := otlpbridge.NewCMAdapterPublisher(cm)
143148

144-
// Initialize main OTLP bridge on port 4317
149+
// Initialize main OTLP bridge - this is critical, so failure stops initialization
145150
bridgeConfig := otlpbridge.BridgeConfig{
146-
ListenAddr: ":4317",
151+
ListenAddr: ":" + strconv.Itoa(otlpPort),
147152
Encoding: "protobuf",
148153
}
149154
var err error
150155
fleetManager.otlpBridge, err = otlpbridge.NewBridgeServer(bridgeConfig, fleetManager.policyManager.GetRepo(), fleetManager.logger)
151156
if err != nil {
152-
fleetManager.logger.Error("failed to create OTLP bridge", slog.Any("error", err))
157+
fleetManager.logger.Error("failed to create OTLP bridge", slog.Any("error", err), slog.Int("port", otlpPort))
153158
return
154159
}
155160
if err := fleetManager.otlpBridge.Start(context.Background()); err != nil {
156-
fleetManager.logger.Error("failed to start OTLP bridge", slog.Any("error", err))
161+
fleetManager.logger.Error("failed to start OTLP bridge", slog.Any("error", err), slog.Int("port", otlpPort))
162+
// Clean up the bridge instance if start failed
163+
fleetManager.otlpBridge = nil
157164
return
158165
}
159166
fleetManager.otlpBridge.SetPublisher(pub)
160167
fleetManager.otlpBridge.SetTopic(topics.Ingest)
161-
fleetManager.logger.Info("OTLP bridge bound to Fleet MQTT", slog.String("topic", topics.Ingest), slog.String("port", "4317"))
168+
fleetManager.logger.Info("OTLP bridge bound to Fleet MQTT", slog.String("topic", topics.Ingest), slog.Int("port", otlpPort))
162169

163-
// Initialize telemetry OTLP bridge on port 4318
170+
// Initialize telemetry OTLP bridge - this is non-critical, so failure is logged but doesn't stop initialization
164171
telemetryBridgeConfig := otlpbridge.BridgeConfig{
165-
ListenAddr: ":4318",
172+
ListenAddr: ":" + strconv.Itoa(telemetryPort),
166173
Encoding: "protobuf",
167174
}
168175
fleetManager.telemetryBridge, err = otlpbridge.NewBridgeServer(telemetryBridgeConfig, fleetManager.policyManager.GetRepo(), fleetManager.logger)
169176
if err != nil {
170-
fleetManager.logger.Error("failed to create telemetry OTLP bridge", slog.Any("error", err))
177+
fleetManager.logger.Warn("failed to create telemetry OTLP bridge, continuing without it", slog.Any("error", err), slog.Int("port", telemetryPort))
171178
return
172179
}
173180
if err := fleetManager.telemetryBridge.Start(context.Background()); err != nil {
174-
fleetManager.logger.Error("failed to start telemetry OTLP bridge", slog.Any("error", err))
181+
fleetManager.logger.Warn("failed to start telemetry OTLP bridge, continuing without it", slog.Any("error", err), slog.Int("port", telemetryPort))
182+
// Clean up the bridge instance if start failed
183+
fleetManager.telemetryBridge = nil
175184
return
176185
}
177186
telemetryTopic := topics.Ingest + "/telemetry"
178187
fleetManager.telemetryBridge.SetPublisher(pub)
179188
fleetManager.telemetryBridge.SetTopic(telemetryTopic)
180-
fleetManager.logger.Info("Telemetry OTLP bridge bound to Fleet MQTT", slog.String("topic", telemetryTopic), slog.String("port", "4318"))
189+
fleetManager.logger.Info("Telemetry OTLP bridge bound to Fleet MQTT", slog.String("topic", telemetryTopic), slog.Int("port", telemetryPort))
181190
})
182191

183192
return nil
184193
}
185194

195+
// getOTLPPort returns the configured port or the default if not set.
196+
func getOTLPPort(configuredPort *int, defaultPort int) int {
197+
if configuredPort != nil && *configuredPort > 0 {
198+
return *configuredPort
199+
}
200+
return defaultPort
201+
}
202+
186203
func (fleetManager *fleetConfigManager) configToSafeString(cfg config.Config) (string, error) {
187204
if cfg.OrbAgent.ConfigManager.Sources.Fleet.ClientSecret != "" {
188205
cfg.OrbAgent.ConfigManager.Sources.Fleet.ClientSecret = "******"

‎agent/otlpbridge/handlers.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ func (s *traceServer) Export(ctx context.Context, req *collectortrace.ExportTrac
2222
if pub == nil {
2323
return nil, fmt.Errorf("publisher not yet initialized")
2424
}
25-
topic := s.bridge.GetIngestTopic()
25+
topic := s.bridge.GetTopic()
2626
if topic == "" {
2727
return nil, fmt.Errorf("topic not yet initialized")
2828
}
@@ -48,7 +48,7 @@ func (s *metricsServer) Export(ctx context.Context, req *collectormetrics.Export
4848
if pub == nil {
4949
return nil, fmt.Errorf("publisher not yet initialized")
5050
}
51-
topic := s.bridge.GetIngestTopic()
51+
topic := s.bridge.GetTopic()
5252
if topic == "" {
5353
return nil, fmt.Errorf("topic not yet initialized")
5454
}
@@ -74,7 +74,7 @@ func (s *logsServer) Export(ctx context.Context, req *collectorlogs.ExportLogsSe
7474
if pub == nil {
7575
return nil, fmt.Errorf("publisher not yet initialized")
7676
}
77-
topic := s.bridge.GetIngestTopic()
77+
topic := s.bridge.GetTopic()
7878
if topic == "" {
7979
return nil, fmt.Errorf("topic not yet initialized")
8080
}

‎agent/otlpbridge/server.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,8 +73,8 @@ func (s *BridgeServer) GetPublisher() Publisher {
7373
return s.publisher
7474
}
7575

76-
// GetIngestTopic returns the current topic (for handlers).
77-
func (s *BridgeServer) GetIngestTopic() string {
76+
// GetTopic returns the current topic (for handlers).
77+
func (s *BridgeServer) GetTopic() string {
7878
s.mu.RLock()
7979
defer s.mu.RUnlock()
8080
return s.ingestTopic

0 commit comments

Comments
 (0)