diff --git a/agent.go b/agent.go index af1af63..0c8cd3b 100644 --- a/agent.go +++ b/agent.go @@ -106,6 +106,7 @@ func runSession(ctx context.Context, baseURL, token string) error { credTestResultCh := make(chan *pb.CredentialTestResult, 5000) monitoringCheckCh := make(chan *pb.MonitoringCheck, 10000) checkResultCh := make(chan *pb.CheckResult, 10000) + lldpTopologyResultCh := make(chan *pb.LldpTopologyResult, 1000) // Ref counter for outbound messages var refCounter atomic.Uint64 @@ -277,7 +278,7 @@ func runSession(ctx context.Context, baseURL, token string) error { slog.Warn("invalid message", "error", err) continue } - if handleMessage(ctx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh) { + if handleMessage(ctx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh) { flushSnmpBatch() return errRestartRequested } @@ -304,6 +305,10 @@ func runSession(ctx context.Context, baseURL, token string) error { sendBinaryResult("check_result", result) slog.Info("sent check result", "check", result.CheckId, "status", result.Status) + case result := <-lldpTopologyResultCh: + sendBinaryResult("lldp_topology_result", result) + slog.Info("sent LLDP topology result", "device", result.DeviceId, "neighbors", len(result.Neighbors)) + case <-flushTicker.C: flushSnmpBatch() @@ -345,6 +350,7 @@ func handleMessage( credTestResultCh chan<- *pb.CredentialTestResult, monitoringCheckCh chan<- *pb.MonitoringCheck, checkResultCh chan<- *pb.CheckResult, + lldpTopologyResultCh chan<- *pb.LldpTopologyResult, ) bool { switch msg.Event { case "phx_reply": @@ -374,7 +380,7 @@ func handleMessage( } slog.Info("received jobs", "count", len(jobList.Jobs)) for _, job := range jobList.Jobs { - dispatchJob(ctx, job, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh) + dispatchJob(ctx, job, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh) } case "check_jobs": @@ -442,6 +448,7 @@ func dispatchJob( credTestResultCh chan<- *pb.CredentialTestResult, monitoringCheckCh chan<- *pb.MonitoringCheck, checkResultCh chan<- *pb.CheckResult, + lldpTopologyResultCh chan<- *pb.LldpTopologyResult, ) { slog.Info("starting job", "job_id", job.JobId, "type", job.JobType) @@ -453,6 +460,8 @@ func dispatchJob( ok = pools.snmp.submit(ctx, func() { executeCredentialTest(ctx, job, credTestResultCh) }) case pb.JobType_PING: ok = pools.ping.submit(ctx, func() { executePingJob(ctx, job, monitoringCheckCh) }) + case pb.JobType_LLDP_TOPOLOGY: + ok = pools.snmp.submit(ctx, func() { executeLldpTopologyJob(ctx, job, lldpTopologyResultCh) }) default: ok = pools.snmp.submit(ctx, func() { executeSnmpJob(ctx, job, snmpResultCh) }) } diff --git a/agent_test.go b/agent_test.go index f77abeb..fb469cc 100644 --- a/agent_test.go +++ b/agent_test.go @@ -108,7 +108,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "phx_reply", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "phx_reply", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Just verify it doesn't panic }) @@ -136,7 +136,7 @@ func TestHandleMessage(t *testing.T) { SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1", Port: 161}, }) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Wait for goroutine to finish select { case <-snmpCh: @@ -151,7 +151,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -162,7 +162,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": "not-base64!!!"}) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) }) t.Run("invalid protobuf", func(t *testing.T) { @@ -172,7 +172,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": base64.StdEncoding.EncodeToString([]byte{0xFF, 0xFF, 0xFF})}) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) }) t.Run("restart", func(t *testing.T) { @@ -181,7 +181,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - shouldEnd := handleMessage(context.Background(), channelMsg{Event: "restart", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + shouldEnd := handleMessage(context.Background(), channelMsg{Event: "restart", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if !shouldEnd { t.Error("expected handleMessage to return true for restart") @@ -204,7 +204,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"url": "https://example.com/agent", "checksum": "abc123"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if calledURL != "https://example.com/agent" { t.Errorf("expected update URL %q, got %q", "https://example.com/agent", calledURL) @@ -228,7 +228,7 @@ func TestHandleMessage(t *testing.T) { checkCh := make(chan *pb.CheckResult, 1) // Missing URL field payload, _ := json.Marshal(map[string]string{"checksum": "abc123"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if called { t.Error("selfUpdate should not be called with empty URL") @@ -249,7 +249,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"url": "https://example.com/agent", "checksum": "abc123"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -269,7 +269,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"url": "https://example.com/agent"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if called { t.Error("selfUpdate should not be called with empty checksum") @@ -282,7 +282,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "some_unknown_event", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "some_unknown_event", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should just log and not panic }) @@ -299,7 +299,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case <-checkCh: case <-time.After(5 * time.Second): @@ -313,7 +313,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -324,7 +324,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": "not-base64!!!"}) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -335,7 +335,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": base64.StdEncoding.EncodeToString([]byte{0xFF, 0xFF, 0xFF})}) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -362,7 +362,7 @@ func TestHandleMessage(t *testing.T) { JobType: pb.JobType_DISCOVER, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, }) - handleMessage(context.Background(), channelMsg{Event: "discovery_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "discovery_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case <-snmpCh: case <-time.After(2 * time.Second): @@ -390,7 +390,7 @@ func TestHandleMessage(t *testing.T) { JobType: pb.JobType_MIKROTIK, MikrotikDevice: &pb.MikrotikDevice{Ip: "10.0.0.1", SshPort: 22, Username: "admin", Password: "pass"}, }) - handleMessage(context.Background(), channelMsg{Event: "backup_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "backup_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case result := <-mtCh: if result.Error != "" { @@ -414,7 +414,7 @@ func TestHandleMessageRejectsOversizedPayload(t *testing.T) { encoded := base64.StdEncoding.EncodeToString(oversized) payload, _ := json.Marshal(map[string]string{"binary": encoded}) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Verify no jobs were dispatched select { @@ -518,7 +518,7 @@ func TestDispatchJob(t *testing.T) { JobId: "mt1", JobType: pb.JobType_MIKROTIK, MikrotikDevice: &pb.MikrotikDevice{Ip: "10.0.0.1", Port: 8728}, - }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case result := <-mtCh: @@ -547,7 +547,7 @@ func TestDispatchJob(t *testing.T) { JobId: "tc1", JobType: pb.JobType_TEST_CREDENTIALS, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, - }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case result := <-credCh: @@ -576,7 +576,7 @@ func TestDispatchJob(t *testing.T) { JobId: "p1", JobType: pb.JobType_PING, SnmpDevice: &pb.SnmpDevice{Ip: "127.0.0.1"}, - }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case result := <-monCh: @@ -609,7 +609,7 @@ func TestDispatchJob(t *testing.T) { JobId: "s1", JobType: pb.JobType_POLL, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, - }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh) + }, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case <-snmpCh: @@ -792,19 +792,19 @@ func TestDispatchJobCancelledContext(t *testing.T) { // All dispatch types should hit the "pool full" warning dispatchJob(ctx, &pb.AgentJob{ JobId: "s1", JobType: pb.JobType_POLL, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, - }, p, snmpCh, mtCh, credCh, monCh, checkCh) + }, p, snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) dispatchJob(ctx, &pb.AgentJob{ JobId: "m1", JobType: pb.JobType_MIKROTIK, MikrotikDevice: &pb.MikrotikDevice{Ip: "10.0.0.1"}, - }, p, snmpCh, mtCh, credCh, monCh, checkCh) + }, p, snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) dispatchJob(ctx, &pb.AgentJob{ JobId: "tc1", JobType: pb.JobType_TEST_CREDENTIALS, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, - }, p, snmpCh, mtCh, credCh, monCh, checkCh) + }, p, snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) dispatchJob(ctx, &pb.AgentJob{ JobId: "p1", JobType: pb.JobType_PING, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, - }, p, snmpCh, mtCh, credCh, monCh, checkCh) + }, p, snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) close(done) } diff --git a/lldp.go b/lldp.go new file mode 100644 index 0000000..0e6b795 --- /dev/null +++ b/lldp.go @@ -0,0 +1,249 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "net" + "strconv" + "strings" + "time" + + "github.com/gosnmp/gosnmp" + "github.com/towerops-app/towerops-agent/pb" +) + +// LLDP-MIB OIDs (IEEE 802.1AB) +const ( + oidLocSysName = "1.0.8802.1.1.2.1.3.3.0" + oidLocPortDesc = "1.0.8802.1.1.2.1.3.7.1.4" + oidRemPortId = "1.0.8802.1.1.2.1.4.1.1.7" + oidRemPortDesc = "1.0.8802.1.1.2.1.4.1.1.8" + oidRemSysName = "1.0.8802.1.1.2.1.4.1.1.9" + oidRemManAddr = "1.0.8802.1.1.2.1.4.2.1.3" +) + +// executeLldpTopologyJob performs LLDP neighbor discovery via SNMP. +func executeLldpTopologyJob(ctx context.Context, job *pb.AgentJob, resultCh chan<- *pb.LldpTopologyResult) { + deviceID := job.DeviceId + jobID := job.JobId + + if job.SnmpDevice == nil { + slog.Error("missing SNMP config for LLDP job", "job_id", jobID, "device_id", deviceID) + return + } + + snmpDev := job.SnmpDevice + client, err := newSnmpConn(snmpDev) + if err != nil { + slog.Error("failed to create SNMP client for LLDP", "job_id", jobID, "device_id", deviceID, "error", err) + return + } + defer func() { + if err := client.Conn.Close(); err != nil { + slog.Debug("SNMP close error", "error", err) + } + }() + + if err := client.Connect(); err != nil { + slog.Error("SNMP connect failed for LLDP", "job_id", jobID, "device_id", deviceID, "error", err) + return + } + + result, err := discoverLldpNeighbors(client, deviceID, jobID) + if err != nil { + slog.Error("LLDP discovery failed", "job_id", jobID, "device_id", deviceID, "error", err) + return + } + + select { + case resultCh <- result: + slog.Info("LLDP topology discovered", "job_id", jobID, "device_id", deviceID, "neighbors", len(result.Neighbors)) + case <-ctx.Done(): + slog.Warn("LLDP result send cancelled", "job_id", jobID) + } +} + +// discoverLldpNeighbors walks LLDP-MIB tables and returns discovered neighbors. +func discoverLldpNeighbors(client *gosnmp.GoSNMP, deviceID, jobID string) (*pb.LldpTopologyResult, error) { + now := time.Now().Unix() + result := &pb.LldpTopologyResult{ + DeviceId: deviceID, + JobId: jobID, + Timestamp: now, + } + + // Get local system name + sysNamePkt, err := client.Get([]string{oidLocSysName}) + if err == nil && len(sysNamePkt.Variables) > 0 { + result.LocalSystemName = snmpValueToString(sysNamePkt.Variables[0]) + } + + // Walk local port descriptions (indexed by port number) + localPorts := make(map[string]string) + if err := client.Walk(oidLocPortDesc, func(pdu gosnmp.SnmpPDU) error { + portNum := extractSuffix(pdu.Name, oidLocPortDesc) + if portNum != "" { + localPorts[portNum] = snmpValueToString(pdu) + } + return nil + }); err != nil { + slog.Warn("failed to walk local ports", "error", err) + } + + // Walk remote system names (indexed by timeMark.portNum.remIndex) + sysNames := make(map[string]string) + if err := client.Walk(oidRemSysName, func(pdu gosnmp.SnmpPDU) error { + key := parseRemoteKey(pdu.Name, oidRemSysName) + if key != "" { + sysNames[key] = snmpValueToString(pdu) + } + return nil + }); err != nil { + slog.Warn("failed to walk remote sys names", "error", err) + return result, nil // Return empty result, not an error + } + + // If no neighbors found, return early + if len(sysNames) == 0 { + return result, nil + } + + // Walk remote port descriptions + remotePorts := make(map[string]string) + _ = client.Walk(oidRemPortDesc, func(pdu gosnmp.SnmpPDU) error { + key := parseRemoteKey(pdu.Name, oidRemPortDesc) + if key != "" { + remotePorts[key] = snmpValueToString(pdu) + } + return nil + }) + + // Walk remote port IDs (fallback when description is empty) + remotePortIds := make(map[string]string) + _ = client.Walk(oidRemPortId, func(pdu gosnmp.SnmpPDU) error { + key := parseRemoteKey(pdu.Name, oidRemPortId) + if key != "" { + remotePortIds[key] = snmpValueToString(pdu) + } + return nil + }) + + // Walk management addresses (indexed by timeMark.portNum.remIndex.addrSubtype.addrLen.addr[bytes]) + mgmtAddrs := make(map[string][]string) + _ = client.Walk(oidRemManAddr, func(pdu gosnmp.SnmpPDU) error { + key, ip := parseMgmtAddr(pdu.Name) + if key != "" && ip != "" { + mgmtAddrs[key] = append(mgmtAddrs[key], ip) + } + return nil + }) + + // Assemble neighbor list + for key, neighborName := range sysNames { + if neighborName == "" { + continue + } + + parts := strings.Split(key, ".") + if len(parts) < 2 { + continue + } + portNum := parts[1] // timeMark.portNum.remIndex -> parts[1] is portNum + + localPort := localPorts[portNum] + if localPort == "" { + localPort = "port-" + portNum + } + + neighbor := &pb.LldpNeighbor{ + NeighborName: neighborName, + LocalPort: localPort, + RemotePort: remotePorts[key], + RemotePortId: remotePortIds[key], + ManagementAddresses: mgmtAddrs[key], + } + + result.Neighbors = append(result.Neighbors, neighbor) + } + + return result, nil +} + +// extractSuffix strips the base OID prefix and returns the suffix. +func extractSuffix(oid, base string) string { + prefix := "." + base + "." + if strings.HasPrefix(oid, prefix) { + return strings.TrimPrefix(oid, prefix) + } + // Try without leading dot on oid + prefix = base + "." + if strings.HasPrefix(oid, prefix) { + return strings.TrimPrefix(oid, prefix) + } + return "" +} + +// parseRemoteKey extracts a remote table key from OID: timeMark.portNum.remIndex +func parseRemoteKey(oid, base string) string { + suffix := extractSuffix(oid, base) + if suffix == "" { + return "" + } + parts := strings.SplitN(suffix, ".", 4) + if len(parts) < 3 { + return "" + } + // Return full key as string: timeMark.portNum.remIndex + return parts[0] + "." + parts[1] + "." + parts[2] +} + +// parseMgmtAddr parses a management address OID. +// Format: timeMark.portNum.remIndex.addrSubtype.addrLen.addr[bytes] +// addrSubtype 1 = IPv4 (4 bytes), 2 = IPv6 (16 bytes) +func parseMgmtAddr(oid string) (key string, ip string) { + suffix := extractSuffix(oid, oidRemManAddr) + if suffix == "" { + return "", "" + } + + parts := strings.Split(suffix, ".") + // Minimum: timeMark(1) portNum(1) remIndex(1) addrSubtype(1) addrLen(1) addr(>=4) + if len(parts) < 9 { + return "", "" + } + + // timeMark.portNum.remIndex + key = parts[0] + "." + parts[1] + "." + parts[2] + addrSubtype := parts[3] + // parts[4] is addrLen (we trust the subtype to determine length) + + switch addrSubtype { + case "1": // IPv4 + if len(parts) < 9 { + return key, "" + } + ip = strings.Join(parts[5:9], ".") + case "2": // IPv6 + if len(parts) < 21 { + return key, "" + } + // Convert 16 octets to IPv6 hex format + var ipv6Parts []string + for i := 0; i < 16; i += 2 { + a, _ := strconv.Atoi(parts[5+i]) + b, _ := strconv.Atoi(parts[5+i+1]) + ipv6Parts = append(ipv6Parts, fmt.Sprintf("%x", a*256+b)) + } + ip = strings.Join(ipv6Parts, ":") + default: + return key, "" + } + + // Validate IP address + if net.ParseIP(ip) == nil { + return key, "" + } + + return key, ip +} diff --git a/pb/agent.pb.go b/pb/agent.pb.go index 26aebd1..e3a5c88 100644 --- a/pb/agent.pb.go +++ b/pb/agent.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v6.32.1 +// protoc v7.34.0 // source: proto/agent.proto package pb @@ -29,6 +29,7 @@ const ( JobType_MIKROTIK JobType = 2 JobType_TEST_CREDENTIALS JobType = 3 JobType_PING JobType = 4 + JobType_LLDP_TOPOLOGY JobType = 5 // Discover all LLDP neighbors for topology mapping ) // Enum value maps for JobType. @@ -39,6 +40,7 @@ var ( 2: "MIKROTIK", 3: "TEST_CREDENTIALS", 4: "PING", + 5: "LLDP_TOPOLOGY", } JobType_value = map[string]int32{ "DISCOVER": 0, @@ -46,6 +48,7 @@ var ( "MIKROTIK": 2, "TEST_CREDENTIALS": 3, "PING": 4, + "LLDP_TOPOLOGY": 5, } ) @@ -2512,6 +2515,158 @@ func (x *MikrotikSentence) GetAttributes() map[string]string { return nil } +type LldpTopologyResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + DeviceId string `protobuf:"bytes,1,opt,name=device_id,json=deviceId,proto3" json:"device_id,omitempty"` + JobId string `protobuf:"bytes,2,opt,name=job_id,json=jobId,proto3" json:"job_id,omitempty"` + LocalSystemName string `protobuf:"bytes,3,opt,name=local_system_name,json=localSystemName,proto3" json:"local_system_name,omitempty"` + Neighbors []*LldpNeighbor `protobuf:"bytes,4,rep,name=neighbors,proto3" json:"neighbors,omitempty"` + Timestamp int64 `protobuf:"varint,5,opt,name=timestamp,proto3" json:"timestamp,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LldpTopologyResult) Reset() { + *x = LldpTopologyResult{} + mi := &file_proto_agent_proto_msgTypes[31] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LldpTopologyResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LldpTopologyResult) ProtoMessage() {} + +func (x *LldpTopologyResult) ProtoReflect() protoreflect.Message { + mi := &file_proto_agent_proto_msgTypes[31] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LldpTopologyResult.ProtoReflect.Descriptor instead. +func (*LldpTopologyResult) Descriptor() ([]byte, []int) { + return file_proto_agent_proto_rawDescGZIP(), []int{31} +} + +func (x *LldpTopologyResult) GetDeviceId() string { + if x != nil { + return x.DeviceId + } + return "" +} + +func (x *LldpTopologyResult) GetJobId() string { + if x != nil { + return x.JobId + } + return "" +} + +func (x *LldpTopologyResult) GetLocalSystemName() string { + if x != nil { + return x.LocalSystemName + } + return "" +} + +func (x *LldpTopologyResult) GetNeighbors() []*LldpNeighbor { + if x != nil { + return x.Neighbors + } + return nil +} + +func (x *LldpTopologyResult) GetTimestamp() int64 { + if x != nil { + return x.Timestamp + } + return 0 +} + +type LldpNeighbor struct { + state protoimpl.MessageState `protogen:"open.v1"` + NeighborName string `protobuf:"bytes,1,opt,name=neighbor_name,json=neighborName,proto3" json:"neighbor_name,omitempty"` + LocalPort string `protobuf:"bytes,2,opt,name=local_port,json=localPort,proto3" json:"local_port,omitempty"` + RemotePort string `protobuf:"bytes,3,opt,name=remote_port,json=remotePort,proto3" json:"remote_port,omitempty"` + RemotePortId string `protobuf:"bytes,4,opt,name=remote_port_id,json=remotePortId,proto3" json:"remote_port_id,omitempty"` + ManagementAddresses []string `protobuf:"bytes,5,rep,name=management_addresses,json=managementAddresses,proto3" json:"management_addresses,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LldpNeighbor) Reset() { + *x = LldpNeighbor{} + mi := &file_proto_agent_proto_msgTypes[32] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LldpNeighbor) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LldpNeighbor) ProtoMessage() {} + +func (x *LldpNeighbor) ProtoReflect() protoreflect.Message { + mi := &file_proto_agent_proto_msgTypes[32] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LldpNeighbor.ProtoReflect.Descriptor instead. +func (*LldpNeighbor) Descriptor() ([]byte, []int) { + return file_proto_agent_proto_rawDescGZIP(), []int{32} +} + +func (x *LldpNeighbor) GetNeighborName() string { + if x != nil { + return x.NeighborName + } + return "" +} + +func (x *LldpNeighbor) GetLocalPort() string { + if x != nil { + return x.LocalPort + } + return "" +} + +func (x *LldpNeighbor) GetRemotePort() string { + if x != nil { + return x.RemotePort + } + return "" +} + +func (x *LldpNeighbor) GetRemotePortId() string { + if x != nil { + return x.RemotePortId + } + return "" +} + +func (x *LldpNeighbor) GetManagementAddresses() []string { + if x != nil { + return x.ManagementAddresses + } + return nil +} + var File_proto_agent_proto protoreflect.FileDescriptor const file_proto_agent_proto_rawDesc = "" + @@ -2733,13 +2888,28 @@ const file_proto_agent_proto_rawDesc = "" + "attributes\x1a=\n" + "\x0fAttributesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01*O\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xce\x01\n" + + "\x12LldpTopologyResult\x12\x1b\n" + + "\tdevice_id\x18\x01 \x01(\tR\bdeviceId\x12\x15\n" + + "\x06job_id\x18\x02 \x01(\tR\x05jobId\x12*\n" + + "\x11local_system_name\x18\x03 \x01(\tR\x0flocalSystemName\x12:\n" + + "\tneighbors\x18\x04 \x03(\v2\x1c.towerops.agent.LldpNeighborR\tneighbors\x12\x1c\n" + + "\ttimestamp\x18\x05 \x01(\x03R\ttimestamp\"\xcc\x01\n" + + "\fLldpNeighbor\x12#\n" + + "\rneighbor_name\x18\x01 \x01(\tR\fneighborName\x12\x1d\n" + + "\n" + + "local_port\x18\x02 \x01(\tR\tlocalPort\x12\x1f\n" + + "\vremote_port\x18\x03 \x01(\tR\n" + + "remotePort\x12$\n" + + "\x0eremote_port_id\x18\x04 \x01(\tR\fremotePortId\x121\n" + + "\x14management_addresses\x18\x05 \x03(\tR\x13managementAddresses*b\n" + "\aJobType\x12\f\n" + "\bDISCOVER\x10\x00\x12\b\n" + "\x04POLL\x10\x01\x12\f\n" + "\bMIKROTIK\x10\x02\x12\x14\n" + "\x10TEST_CREDENTIALS\x10\x03\x12\b\n" + - "\x04PING\x10\x04*\x1e\n" + + "\x04PING\x10\x04\x12\x11\n" + + "\rLLDP_TOPOLOGY\x10\x05*\x1e\n" + "\tQueryType\x12\a\n" + "\x03GET\x10\x00\x12\b\n" + "\x04WALK\x10\x01B+Z)github.com/towerops-app/towerops-agent/pbb\x06proto3" @@ -2757,7 +2927,7 @@ func file_proto_agent_proto_rawDescGZIP() []byte { } var file_proto_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_proto_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 36) +var file_proto_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 38) var file_proto_agent_proto_goTypes = []any{ (JobType)(0), // 0: towerops.agent.JobType (QueryType)(0), // 1: towerops.agent.QueryType @@ -2792,11 +2962,13 @@ var file_proto_agent_proto_goTypes = []any{ (*MikrotikCommand)(nil), // 30: towerops.agent.MikrotikCommand (*MikrotikResult)(nil), // 31: towerops.agent.MikrotikResult (*MikrotikSentence)(nil), // 32: towerops.agent.MikrotikSentence - nil, // 33: towerops.agent.Sensor.MetadataEntry - nil, // 34: towerops.agent.HttpCheckConfig.HeadersEntry - nil, // 35: towerops.agent.SnmpResult.OidValuesEntry - nil, // 36: towerops.agent.MikrotikCommand.ArgsEntry - nil, // 37: towerops.agent.MikrotikSentence.AttributesEntry + (*LldpTopologyResult)(nil), // 33: towerops.agent.LldpTopologyResult + (*LldpNeighbor)(nil), // 34: towerops.agent.LldpNeighbor + nil, // 35: towerops.agent.Sensor.MetadataEntry + nil, // 36: towerops.agent.HttpCheckConfig.HeadersEntry + nil, // 37: towerops.agent.SnmpResult.OidValuesEntry + nil, // 38: towerops.agent.MikrotikCommand.ArgsEntry + nil, // 39: towerops.agent.MikrotikSentence.AttributesEntry } var file_proto_agent_proto_depIdxs = []int32{ 3, // 0: towerops.agent.AgentConfig.devices:type_name -> towerops.agent.Device @@ -2804,7 +2976,7 @@ var file_proto_agent_proto_depIdxs = []int32{ 4, // 2: towerops.agent.Device.snmp:type_name -> towerops.agent.SnmpConfig 5, // 3: towerops.agent.Device.sensors:type_name -> towerops.agent.Sensor 6, // 4: towerops.agent.Device.interfaces:type_name -> towerops.agent.Interface - 33, // 5: towerops.agent.Sensor.metadata:type_name -> towerops.agent.Sensor.MetadataEntry + 35, // 5: towerops.agent.Sensor.metadata:type_name -> towerops.agent.Sensor.MetadataEntry 8, // 6: towerops.agent.MetricBatch.metrics:type_name -> towerops.agent.Metric 9, // 7: towerops.agent.Metric.sensor_reading:type_name -> towerops.agent.SensorReading 10, // 8: towerops.agent.Metric.interface_stat:type_name -> towerops.agent.InterfaceStat @@ -2814,7 +2986,7 @@ var file_proto_agent_proto_depIdxs = []int32{ 14, // 12: towerops.agent.Check.http:type_name -> towerops.agent.HttpCheckConfig 15, // 13: towerops.agent.Check.tcp:type_name -> towerops.agent.TcpCheckConfig 16, // 14: towerops.agent.Check.dns:type_name -> towerops.agent.DnsCheckConfig - 34, // 15: towerops.agent.HttpCheckConfig.headers:type_name -> towerops.agent.HttpCheckConfig.HeadersEntry + 36, // 15: towerops.agent.HttpCheckConfig.headers:type_name -> towerops.agent.HttpCheckConfig.HeadersEntry 13, // 16: towerops.agent.CheckList.checks:type_name -> towerops.agent.Check 22, // 17: towerops.agent.AgentJobList.jobs:type_name -> towerops.agent.AgentJob 0, // 18: towerops.agent.AgentJob.job_type:type_name -> towerops.agent.JobType @@ -2824,15 +2996,16 @@ var file_proto_agent_proto_depIdxs = []int32{ 30, // 22: towerops.agent.AgentJob.mikrotik_commands:type_name -> towerops.agent.MikrotikCommand 1, // 23: towerops.agent.SnmpQuery.query_type:type_name -> towerops.agent.QueryType 0, // 24: towerops.agent.SnmpResult.job_type:type_name -> towerops.agent.JobType - 35, // 25: towerops.agent.SnmpResult.oid_values:type_name -> towerops.agent.SnmpResult.OidValuesEntry - 36, // 26: towerops.agent.MikrotikCommand.args:type_name -> towerops.agent.MikrotikCommand.ArgsEntry + 37, // 25: towerops.agent.SnmpResult.oid_values:type_name -> towerops.agent.SnmpResult.OidValuesEntry + 38, // 26: towerops.agent.MikrotikCommand.args:type_name -> towerops.agent.MikrotikCommand.ArgsEntry 32, // 27: towerops.agent.MikrotikResult.sentences:type_name -> towerops.agent.MikrotikSentence - 37, // 28: towerops.agent.MikrotikSentence.attributes:type_name -> towerops.agent.MikrotikSentence.AttributesEntry - 29, // [29:29] is the sub-list for method output_type - 29, // [29:29] is the sub-list for method input_type - 29, // [29:29] is the sub-list for extension type_name - 29, // [29:29] is the sub-list for extension extendee - 0, // [0:29] is the sub-list for field type_name + 39, // 28: towerops.agent.MikrotikSentence.attributes:type_name -> towerops.agent.MikrotikSentence.AttributesEntry + 34, // 29: towerops.agent.LldpTopologyResult.neighbors:type_name -> towerops.agent.LldpNeighbor + 30, // [30:30] is the sub-list for method output_type + 30, // [30:30] is the sub-list for method input_type + 30, // [30:30] is the sub-list for extension type_name + 30, // [30:30] is the sub-list for extension extendee + 0, // [0:30] is the sub-list for field type_name } func init() { file_proto_agent_proto_init() } @@ -2858,7 +3031,7 @@ func file_proto_agent_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_agent_proto_rawDesc), len(file_proto_agent_proto_rawDesc)), NumEnums: 2, - NumMessages: 36, + NumMessages: 38, NumExtensions: 0, NumServices: 0, }, diff --git a/proto/agent.proto b/proto/agent.proto index 955c482..4d74ba7 100644 --- a/proto/agent.proto +++ b/proto/agent.proto @@ -173,6 +173,7 @@ enum JobType { MIKROTIK = 2; TEST_CREDENTIALS = 3; PING = 4; + LLDP_TOPOLOGY = 5; // Discover all LLDP neighbors for topology mapping } enum QueryType { @@ -273,3 +274,21 @@ message MikrotikResult { message MikrotikSentence { map attributes = 1; } + +// LLDP Topology Discovery Results + +message LldpTopologyResult { + string device_id = 1; + string job_id = 2; + string local_system_name = 3; + repeated LldpNeighbor neighbors = 4; + int64 timestamp = 5; +} + +message LldpNeighbor { + string neighbor_name = 1; + string local_port = 2; + string remote_port = 3; + string remote_port_id = 4; + repeated string management_addresses = 5; +}