mchus/logpile
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 20 Wiki Activity

Add pluggable live collectors and simplify API connect form

Browse Source
This commit is contained in:
Mikhail Chusavitin
2026-02-04 19:00:03 +03:00
parent 60c52b18b1
commit c89ee0118f
15 changed files with 939 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
internal/collector/helpers.go Normal file
View File

@@ -0,0 +1,18 @@
package collector
import (
"context"
"time"
)
func sleepWithContext(ctx context.Context, d time.Duration) bool {
timer := time.NewTimer(d)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}

42
internal/collector/ipmi_mock.go Normal file
View File

@@ -0,0 +1,42 @@
package collector
import (
"context"
"time"
"git.mchus.pro/mchus/logpile/internal/models"
)
type IPMIMockConnector struct{}
func NewIPMIMockConnector() *IPMIMockConnector {
return &IPMIMockConnector{}
}
func (c *IPMIMockConnector) Protocol() string {
return "ipmi"
}
func (c *IPMIMockConnector) Collect(ctx context.Context, req Request, emit ProgressFn) (*models.AnalysisResult, error) {
steps := []Progress{
{Status: "running", Progress: 20, Message: "IPMI: подключение к BMC..."},
{Status: "running", Progress: 55, Message: "IPMI: чтение инвентаря..."},
{Status: "running", Progress: 85, Message: "IPMI: нормализация данных..."},
}
for _, step := range steps {
if !sleepWithContext(ctx, 150*time.Millisecond) {
return nil, ctx.Err()
}
if emit != nil {
emit(step)
}
}
return &models.AnalysisResult{
Events: make([]models.Event, 0),
FRU: make([]models.FRUInfo, 0),
Sensors: make([]models.SensorReading, 0),
Hardware: &models.HardwareConfig{},
}, nil
}

414
internal/collector/redfish.go Normal file
View File

@@ -0,0 +1,414 @@
package collector
import (
"context"
"crypto/tls"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"path"
"strconv"
"strings"
"time"
"git.mchus.pro/mchus/logpile/internal/models"
)
type RedfishConnector struct {
timeout time.Duration
}
func NewRedfishConnector() *RedfishConnector {
return &RedfishConnector{
timeout: 10 * time.Second,
}
}
func (c *RedfishConnector) Protocol() string {
return "redfish"
}
func (c *RedfishConnector) Collect(ctx context.Context, req Request, emit ProgressFn) (*models.AnalysisResult, error) {
baseURL, err := c.baseURL(req)
if err != nil {
return nil, err
}
client := c.httpClient(req)
if emit != nil {
emit(Progress{Status: "running", Progress: 10, Message: "Redfish: подключение к BMC..."})
}
if _, err := c.getJSON(ctx, client, req, baseURL, "/redfish/v1"); err != nil {
return nil, fmt.Errorf("redfish service root: %w", err)
}
if emit != nil {
emit(Progress{Status: "running", Progress: 30, Message: "Redfish: чтение данных системы..."})
}
systemDoc, err := c.getJSON(ctx, client, req, baseURL, "/redfish/v1/Systems/1")
if err != nil {
return nil, fmt.Errorf("system info: %w", err)
}
biosDoc, _ := c.getJSON(ctx, client, req, baseURL, "/redfish/v1/Systems/1/Bios")
secureBootDoc, _ := c.getJSON(ctx, client, req, baseURL, "/redfish/v1/Systems/1/SecureBoot")
if emit != nil {
emit(Progress{Status: "running", Progress: 55, Message: "Redfish: чтение CPU/RAM/Storage..."})
}
processors, _ := c.getCollectionMembers(ctx, client, req, baseURL, "/redfish/v1/Systems/1/Processors")
memory, _ := c.getCollectionMembers(ctx, client, req, baseURL, "/redfish/v1/Systems/1/Memory")
storageMembers, _ := c.getCollectionMembers(ctx, client, req, baseURL, "/redfish/v1/Systems/1/Storage")
storageDevices := c.collectStorage(ctx, client, req, baseURL, storageMembers)
if emit != nil {
emit(Progress{Status: "running", Progress: 80, Message: "Redfish: чтение сетевых и BMC настроек..."})
}
nics := c.collectNICs(ctx, client, req, baseURL)
managerDoc, _ := c.getJSON(ctx, client, req, baseURL, "/redfish/v1/Managers/1")
networkProtocolDoc, _ := c.getJSON(ctx, client, req, baseURL, "/redfish/v1/Managers/1/NetworkProtocol")
result := &models.AnalysisResult{
Events: make([]models.Event, 0),
FRU: make([]models.FRUInfo, 0),
Sensors: make([]models.SensorReading, 0),
Hardware: &models.HardwareConfig{
BoardInfo: parseBoardInfo(systemDoc),
CPUs: parseCPUs(processors),
Memory: parseMemory(memory),
Storage: storageDevices,
NetworkAdapters: nics,
Firmware: parseFirmware(systemDoc, biosDoc, managerDoc, secureBootDoc, networkProtocolDoc),
},
}
return result, nil
}
func (c *RedfishConnector) httpClient(req Request) *http.Client {
transport := &http.Transport{}
if req.TLSMode == "insecure" {
transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec
}
return &http.Client{
Transport: transport,
Timeout: c.timeout,
}
}
func (c *RedfishConnector) baseURL(req Request) (string, error) {
host := strings.TrimSpace(req.Host)
if host == "" {
return "", fmt.Errorf("empty host")
}
if strings.HasPrefix(host, "http://") || strings.HasPrefix(host, "https://") {
u, err := url.Parse(host)
if err != nil {
return "", fmt.Errorf("invalid host URL: %w", err)
}
u.Path = ""
u.RawQuery = ""
u.Fragment = ""
return strings.TrimRight(u.String(), "/"), nil
}
scheme := "https"
if req.TLSMode == "insecure" && req.Port == 80 {
scheme = "http"
}
return fmt.Sprintf("%s://%s:%d", scheme, host, req.Port), nil
}
func (c *RedfishConnector) collectStorage(ctx context.Context, client *http.Client, req Request, baseURL string, storageMembers []map[string]interface{}) []models.Storage {
var out []models.Storage
for _, member := range storageMembers {
drives, ok := member["Drives"].([]interface{})
if !ok {
continue
}
for _, driveAny := range drives {
driveRef, ok := driveAny.(map[string]interface{})
if !ok {
continue
}
odata := asString(driveRef["@odata.id"])
if odata == "" {
continue
}
driveDoc, err := c.getJSON(ctx, client, req, baseURL, odata)
if err != nil {
continue
}
out = append(out, parseDrive(driveDoc))
}
}
return out
}
func (c *RedfishConnector) collectNICs(ctx context.Context, client *http.Client, req Request, baseURL string) []models.NetworkAdapter {
adapterDocs, err := c.getCollectionMembers(ctx, client, req, baseURL, "/redfish/v1/Chassis/1/NetworkAdapters")
if err != nil {
return nil
}
nics := make([]models.NetworkAdapter, 0, len(adapterDocs))
for _, doc := range adapterDocs {
nics = append(nics, parseNIC(doc))
}
return nics
}
func (c *RedfishConnector) getCollectionMembers(ctx context.Context, client *http.Client, req Request, baseURL, collectionPath string) ([]map[string]interface{}, error) {
collection, err := c.getJSON(ctx, client, req, baseURL, collectionPath)
if err != nil {
return nil, err
}
refs, ok := collection["Members"].([]interface{})
if !ok || len(refs) == 0 {
return []map[string]interface{}{}, nil
}
out := make([]map[string]interface{}, 0, len(refs))
for _, refAny := range refs {
ref, ok := refAny.(map[string]interface{})
if !ok {
continue
}
memberPath := asString(ref["@odata.id"])
if memberPath == "" {
continue
}
memberDoc, err := c.getJSON(ctx, client, req, baseURL, memberPath)
if err != nil {
continue
}
out = append(out, memberDoc)
}
return out, nil
}
func (c *RedfishConnector) getJSON(ctx context.Context, client *http.Client, req Request, baseURL, requestPath string) (map[string]interface{}, error) {
rel := requestPath
if rel == "" {
rel = "/"
}
if !strings.HasPrefix(rel, "/") {
rel = "/" + rel
}
u, err := url.Parse(baseURL)
if err != nil {
return nil, err
}
u.Path = path.Join(strings.TrimSuffix(u.Path, "/"), rel)
httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
if err != nil {
return nil, err
}
httpReq.Header.Set("Accept", "application/json")
switch req.AuthType {
case "password":
httpReq.SetBasicAuth(req.Username, req.Password)
case "token":
httpReq.Header.Set("X-Auth-Token", req.Token)
httpReq.Header.Set("Authorization", "Bearer "+req.Token)
}
resp, err := client.Do(httpReq)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
return nil, fmt.Errorf("status %d from %s: %s", resp.StatusCode, requestPath, strings.TrimSpace(string(body)))
}
var doc map[string]interface{}
dec := json.NewDecoder(resp.Body)
dec.UseNumber()
if err := dec.Decode(&doc); err != nil {
return nil, err
}
return doc, nil
}
func parseBoardInfo(system map[string]interface{}) models.BoardInfo {
return models.BoardInfo{
Manufacturer: asString(system["Manufacturer"]),
ProductName: firstNonEmpty(asString(system["Model"]), asString(system["Name"])),
SerialNumber: asString(system["SerialNumber"]),
PartNumber: asString(system["PartNumber"]),
UUID: asString(system["UUID"]),
}
}
func parseCPUs(docs []map[string]interface{}) []models.CPU {
cpus := make([]models.CPU, 0, len(docs))
for idx, doc := range docs {
cpus = append(cpus, models.CPU{
Socket: idx,
Model: firstNonEmpty(asString(doc["Model"]), asString(doc["Name"])),
Cores: asInt(doc["TotalCores"]),
Threads: asInt(doc["TotalThreads"]),
FrequencyMHz: asInt(doc["OperatingSpeedMHz"]),
MaxFreqMHz: asInt(doc["MaxSpeedMHz"]),
SerialNumber: asString(doc["SerialNumber"]),
})
}
return cpus
}
func parseMemory(docs []map[string]interface{}) []models.MemoryDIMM {
out := make([]models.MemoryDIMM, 0, len(docs))
for _, doc := range docs {
slot := firstNonEmpty(asString(doc["DeviceLocator"]), asString(doc["Name"]), asString(doc["Id"]))
present := true
if strings.EqualFold(asString(doc["Status"]), "Absent") {
present = false
}
if status, ok := doc["Status"].(map[string]interface{}); ok {
state := asString(status["State"])
if strings.EqualFold(state, "Absent") || strings.EqualFold(state, "Disabled") {
present = false
}
}
out = append(out, models.MemoryDIMM{
Slot: slot,
Location: slot,
Present: present,
SizeMB: asInt(doc["CapacityMiB"]),
Type: firstNonEmpty(asString(doc["MemoryDeviceType"]), asString(doc["MemoryType"])),
MaxSpeedMHz: asInt(doc["MaxSpeedMHz"]),
CurrentSpeedMHz: asInt(doc["OperatingSpeedMhz"]),
Manufacturer: asString(doc["Manufacturer"]),
SerialNumber: asString(doc["SerialNumber"]),
PartNumber: asString(doc["PartNumber"]),
Status: mapStatus(doc["Status"]),
})
}
return out
}
func parseDrive(doc map[string]interface{}) models.Storage {
sizeGB := asInt(doc["CapacityBytes"]) / (1024 * 1024 * 1024)
if sizeGB == 0 {
sizeGB = asInt(doc["CapacityGB"])
}
return models.Storage{
Slot: firstNonEmpty(asString(doc["Id"]), asString(doc["Name"])),
Type: firstNonEmpty(asString(doc["MediaType"]), asString(doc["Protocol"])),
Model: firstNonEmpty(asString(doc["Model"]), asString(doc["Name"])),
SizeGB: sizeGB,
SerialNumber: asString(doc["SerialNumber"]),
Manufacturer: asString(doc["Manufacturer"]),
Firmware: asString(doc["Revision"]),
Interface: asString(doc["Protocol"]),
Present: true,
}
}
func parseNIC(doc map[string]interface{}) models.NetworkAdapter {
return models.NetworkAdapter{
Slot: firstNonEmpty(asString(doc["Id"]), asString(doc["Name"])),
Location: asString(doc["Location"]),
Present: !strings.EqualFold(mapStatus(doc["Status"]), "Absent"),
Model: firstNonEmpty(asString(doc["Model"]), asString(doc["Name"])),
Vendor: asString(doc["Manufacturer"]),
SerialNumber: asString(doc["SerialNumber"]),
PartNumber: asString(doc["PartNumber"]),
Status: mapStatus(doc["Status"]),
}
}
func parseFirmware(system, bios, manager, secureBoot, networkProtocol map[string]interface{}) []models.FirmwareInfo {
var out []models.FirmwareInfo
appendFW := func(name, version string) {
version = strings.TrimSpace(version)
if version == "" {
return
}
out = append(out, models.FirmwareInfo{DeviceName: name, Version: version})
}
appendFW("BIOS", asString(system["BiosVersion"]))
appendFW("BIOS", asString(bios["Version"]))
appendFW("BMC", asString(manager["FirmwareVersion"]))
appendFW("SecureBoot", asString(secureBoot["SecureBootCurrentBoot"]))
appendFW("NetworkProtocol", asString(networkProtocol["Id"]))
return out
}
func mapStatus(statusAny interface{}) string {
if statusAny == nil {
return ""
}
if statusMap, ok := statusAny.(map[string]interface{}); ok {
health := asString(statusMap["Health"])
state := asString(statusMap["State"])
return firstNonEmpty(health, state)
}
return asString(statusAny)
}
func asString(v interface{}) string {
switch value := v.(type) {
case nil:
return ""
case string:
return strings.TrimSpace(value)
case json.Number:
return value.String()
default:
return strings.TrimSpace(fmt.Sprintf("%v", value))
}
}
func asInt(v interface{}) int {
switch value := v.(type) {
case nil:
return 0
case int:
return value
case int64:
return int(value)
case float64:
return int(value)
case json.Number:
if i, err := value.Int64(); err == nil {
return int(i)
}
if f, err := value.Float64(); err == nil {
return int(f)
}
case string:
if value == "" {
return 0
}
if i, err := strconv.Atoi(value); err == nil {
return i
}
}
return 0
}
func firstNonEmpty(values ...string) string {
for _, v := range values {
if strings.TrimSpace(v) != "" {
return strings.TrimSpace(v)
}
}
return ""
}

128
internal/collector/redfish_test.go Normal file
View File

@@ -0,0 +1,128 @@
package collector
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
func TestRedfishConnectorCollect(t *testing.T) {
mux := http.NewServeMux()
register := func(path string, payload interface{}) {
mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(payload)
})
}
register("/redfish/v1", map[string]interface{}{"Name": "ServiceRoot"})
register("/redfish/v1/Systems/1", map[string]interface{}{
"Manufacturer": "Supermicro",
"Model": "SYS-TEST",
"SerialNumber": "SYS123",
"BiosVersion": "2.1a",
})
register("/redfish/v1/Systems/1/Bios", map[string]interface{}{"Version": "2.1a"})
register("/redfish/v1/Systems/1/SecureBoot", map[string]interface{}{"SecureBootCurrentBoot": "Enabled"})
register("/redfish/v1/Systems/1/Processors", map[string]interface{}{
"Members": []map[string]string{
{"@odata.id": "/redfish/v1/Systems/1/Processors/CPU1"},
},
})
register("/redfish/v1/Systems/1/Processors/CPU1", map[string]interface{}{
"Name": "CPU1",
"Model": "Xeon Gold",
"TotalCores": 32,
"TotalThreads": 64,
"MaxSpeedMHz": 3600,
})
register("/redfish/v1/Systems/1/Memory", map[string]interface{}{
"Members": []map[string]string{
{"@odata.id": "/redfish/v1/Systems/1/Memory/DIMM1"},
},
})
register("/redfish/v1/Systems/1/Memory/DIMM1", map[string]interface{}{
"Name": "DIMM A1",
"CapacityMiB": 32768,
"MemoryDeviceType": "DDR5",
"OperatingSpeedMhz": 4800,
"Status": map[string]interface{}{
"Health": "OK",
},
})
register("/redfish/v1/Systems/1/Storage", map[string]interface{}{
"Members": []map[string]string{
{"@odata.id": "/redfish/v1/Systems/1/Storage/1"},
},
})
register("/redfish/v1/Systems/1/Storage/1", map[string]interface{}{
"Drives": []map[string]string{
{"@odata.id": "/redfish/v1/Systems/1/Storage/1/Drives/1"},
},
})
register("/redfish/v1/Systems/1/Storage/1/Drives/1", map[string]interface{}{
"Name": "Drive1",
"Model": "NVMe Test",
"MediaType": "SSD",
"Protocol": "NVMe",
"CapacityGB": 960,
"SerialNumber": "SN123",
})
register("/redfish/v1/Chassis/1/NetworkAdapters", map[string]interface{}{
"Members": []map[string]string{
{"@odata.id": "/redfish/v1/Chassis/1/NetworkAdapters/1"},
},
})
register("/redfish/v1/Chassis/1/NetworkAdapters/1", map[string]interface{}{
"Name": "Mellanox",
"Model": "ConnectX-6",
"SerialNumber": "NIC123",
})
register("/redfish/v1/Managers/1", map[string]interface{}{
"FirmwareVersion": "1.25",
})
register("/redfish/v1/Managers/1/NetworkProtocol", map[string]interface{}{
"Id": "NetworkProtocol",
})
ts := httptest.NewServer(mux)
defer ts.Close()
c := NewRedfishConnector()
result, err := c.Collect(context.Background(), Request{
Host: ts.URL,
Port: 443,
Protocol: "redfish",
Username: "admin",
AuthType: "password",
Password: "secret",
TLSMode: "strict",
}, nil)
if err != nil {
t.Fatalf("collect failed: %v", err)
}
if result.Hardware == nil {
t.Fatalf("expected hardware config")
}
if result.Hardware.BoardInfo.ProductName != "SYS-TEST" {
t.Fatalf("unexpected board model: %q", result.Hardware.BoardInfo.ProductName)
}
if len(result.Hardware.CPUs) != 1 {
t.Fatalf("expected one CPU, got %d", len(result.Hardware.CPUs))
}
if len(result.Hardware.Memory) != 1 {
t.Fatalf("expected one DIMM, got %d", len(result.Hardware.Memory))
}
if len(result.Hardware.Storage) != 1 {
t.Fatalf("expected one drive, got %d", len(result.Hardware.Storage))
}
if len(result.Hardware.NetworkAdapters) != 1 {
t.Fatalf("expected one nic, got %d", len(result.Hardware.NetworkAdapters))
}
if len(result.Hardware.Firmware) == 0 {
t.Fatalf("expected firmware entries")
}
}

37
internal/collector/registry.go Normal file
View File

@@ -0,0 +1,37 @@
package collector
import "sync"
type Registry struct {
mu sync.RWMutex
connectors map[string]Connector
}
func NewRegistry() *Registry {
return &Registry{
connectors: make(map[string]Connector),
}
}
func NewDefaultRegistry() *Registry {
r := NewRegistry()
r.Register(NewRedfishConnector())
r.Register(NewIPMIMockConnector())
return r
}
func (r *Registry) Register(connector Connector) {
if connector == nil {
return
}
r.mu.Lock()
r.connectors[connector.Protocol()] = connector
r.mu.Unlock()
}
func (r *Registry) Get(protocol string) (Connector, bool) {
r.mu.RLock()
connector, ok := r.connectors[protocol]
r.mu.RUnlock()
return connector, ok
}

31
internal/collector/types.go Normal file
View File

@@ -0,0 +1,31 @@
package collector
import (
"context"
"git.mchus.pro/mchus/logpile/internal/models"
)
type Request struct {
Host string
Protocol string
Port int
Username string
AuthType string
Password string
Token string
TLSMode string
}
type Progress struct {
Status string
Progress int
Message string
}
type ProgressFn func(Progress)
type Connector interface {
Protocol() string
Collect(ctx context.Context, req Request, emit ProgressFn) (*models.AnalysisResult, error)
}

3
internal/server/collect_handlers_test.go
View File

@@ -14,7 +14,8 @@ import (
func newCollectTestServer() (*Server, *httptest.Server) {
s := &Server{
jobManager: NewJobManager(),
jobManager: NewJobManager(),
collectors: testCollectorRegistry(),
}
mux := http.NewServeMux()
mux.HandleFunc("POST /api/collect", s.handleCollectStart)

63
internal/server/collect_test_helpers_test.go Normal file
View File

@@ -0,0 +1,63 @@
package server
import (
"context"
"strings"
"time"
"git.mchus.pro/mchus/logpile/internal/collector"
"git.mchus.pro/mchus/logpile/internal/models"
)
type mockConnector struct {
protocol string
}
func (c *mockConnector) Protocol() string {
return c.protocol
}
func (c *mockConnector) Collect(ctx context.Context, req collector.Request, emit collector.ProgressFn) (*models.AnalysisResult, error) {
steps := []collector.Progress{
{Status: CollectStatusRunning, Progress: 20, Message: "Подключение..."},
{Status: CollectStatusRunning, Progress: 50, Message: "Сбор инвентаря..."},
{Status: CollectStatusRunning, Progress: 80, Message: "Нормализация..."},
}
for _, step := range steps {
if !collectorSleep(ctx, 100*time.Millisecond) {
return nil, ctx.Err()
}
if emit != nil {
emit(step)
}
}
if strings.Contains(strings.ToLower(req.Host), "fail") {
return nil, context.DeadlineExceeded
}
return &models.AnalysisResult{
Events: make([]models.Event, 0),
FRU: make([]models.FRUInfo, 0),
Sensors: make([]models.SensorReading, 0),
Hardware: &models.HardwareConfig{},
}, nil
}
func testCollectorRegistry() *collector.Registry {
r := collector.NewRegistry()
r.Register(&mockConnector{protocol: "redfish"})
r.Register(&mockConnector{protocol: "ipmi"})
return r
}
func collectorSleep(ctx context.Context, d time.Duration) bool {
timer := time.NewTimer(d)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}

104
internal/server/handlers.go
View File

@@ -13,6 +13,7 @@ import (
"strings"
"time"
"git.mchus.pro/mchus/logpile/internal/collector"
"git.mchus.pro/mchus/logpile/internal/exporter"
"git.mchus.pro/mchus/logpile/internal/models"
"git.mchus.pro/mchus/logpile/internal/parser"
@@ -592,7 +593,7 @@ func (s *Server) handleCollectStart(w http.ResponseWriter, r *http.Request) {
}
job := s.jobManager.CreateJob(req)
s.startMockCollectionJob(job.ID, req)
s.startCollectionJob(job.ID, req)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusAccepted)
@@ -631,7 +632,7 @@ func (s *Server) handleCollectCancel(w http.ResponseWriter, r *http.Request) {
jsonResponse(w, job.toStatusResponse())
}
func (s *Server) startMockCollectionJob(jobID string, req CollectRequest) {
func (s *Server) startCollectionJob(jobID string, req CollectRequest) {
ctx, cancel := context.WithCancel(context.Background())
if attached := s.jobManager.AttachJobCancel(jobID, cancel); !attached {
cancel()
@@ -639,31 +640,37 @@ func (s *Server) startMockCollectionJob(jobID string, req CollectRequest) {
}
go func() {
steps := []struct {
delay time.Duration
status string
progress int
log string
}{
{delay: 250 * time.Millisecond, status: CollectStatusRunning, progress: 20, log: "Подключение..."},
{delay: 250 * time.Millisecond, status: CollectStatusRunning, progress: 50, log: "Сбор инвентаря..."},
{delay: 250 * time.Millisecond, status: CollectStatusRunning, progress: 80, log: "Нормализация..."},
connector, ok := s.getCollector(req.Protocol)
if !ok {
s.jobManager.UpdateJobStatus(jobID, CollectStatusFailed, 100, "Коннектор для протокола не зарегистрирован")
s.jobManager.AppendJobLog(jobID, "Сбор завершен с ошибкой")
return
}
for _, step := range steps {
if !waitWithCancel(ctx, step.delay) {
return
}
emitProgress := func(update collector.Progress) {
if job, ok := s.jobManager.GetJob(jobID); !ok || isTerminalCollectStatus(job.Status) {
return
}
s.jobManager.UpdateJobStatus(jobID, step.status, step.progress, "")
s.jobManager.AppendJobLog(jobID, step.log)
status := update.Status
if status == "" {
status = CollectStatusRunning
}
s.jobManager.UpdateJobStatus(jobID, status, update.Progress, "")
if update.Message != "" {
s.jobManager.AppendJobLog(jobID, update.Message)
}
}
if !waitWithCancel(ctx, 250*time.Millisecond) {
result, err := connector.Collect(ctx, toCollectorRequest(req), emitProgress)
if err != nil {
if ctx.Err() != nil {
return
}
if job, ok := s.jobManager.GetJob(jobID); !ok || isTerminalCollectStatus(job.Status) {
return
}
s.jobManager.UpdateJobStatus(jobID, CollectStatusFailed, 100, err.Error())
s.jobManager.AppendJobLog(jobID, "Сбор завершен с ошибкой")
return
}
@@ -671,31 +678,14 @@ func (s *Server) startMockCollectionJob(jobID string, req CollectRequest) {
return
}
if strings.Contains(strings.ToLower(req.Host), "fail") {
s.jobManager.UpdateJobStatus(jobID, CollectStatusFailed, 100, "Mock: не удалось завершить сбор")
s.jobManager.AppendJobLog(jobID, "Сбор завершен с ошибкой")
return
}
applyCollectSourceMetadata(result, req)
s.jobManager.UpdateJobStatus(jobID, CollectStatusSuccess, 100, "")
s.jobManager.AppendJobLog(jobID, "Сбор завершен")
s.SetResult(newAPIResult(req))
s.SetResult(result)
s.SetDetectedVendor("")
}()
}
func waitWithCancel(ctx context.Context, d time.Duration) bool {
timer := time.NewTimer(d)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}
func validateCollectRequest(req CollectRequest) error {
if strings.TrimSpace(req.Host) == "" {
return fmt.Errorf("field 'host' is required")
@@ -756,16 +746,34 @@ func applyArchiveSourceMetadata(result *models.AnalysisResult) {
result.CollectedAt = time.Now().UTC()
}
func newAPIResult(req CollectRequest) *models.AnalysisResult {
return &models.AnalysisResult{
SourceType: models.SourceTypeAPI,
Protocol: req.Protocol,
TargetHost: req.Host,
CollectedAt: time.Now().UTC(),
Events: make([]models.Event, 0),
FRU: make([]models.FRUInfo, 0),
Sensors: make([]models.SensorReading, 0),
func applyCollectSourceMetadata(result *models.AnalysisResult, req CollectRequest) {
if result == nil {
return
}
result.SourceType = models.SourceTypeAPI
result.Protocol = req.Protocol
result.TargetHost = req.Host
result.CollectedAt = time.Now().UTC()
}
func toCollectorRequest(req CollectRequest) collector.Request {
return collector.Request{
Host: req.Host,
Protocol: req.Protocol,
Port: req.Port,
Username: req.Username,
AuthType: req.AuthType,
Password: req.Password,
Token: req.Token,
TLSMode: req.TLSMode,
}
}
func (s *Server) getCollector(protocol string) (collector.Connector, bool) {
if s.collectors == nil {
s.collectors = collector.NewDefaultRegistry()
}
return s.collectors.Get(protocol)
}
func jsonResponse(w http.ResponseWriter, data interface{}) {

5
internal/server/server.go
View File

@@ -9,6 +9,7 @@ import (
"sync"
"time"
"git.mchus.pro/mchus/logpile/internal/collector"
"git.mchus.pro/mchus/logpile/internal/models"
)
@@ -29,7 +30,8 @@ type Server struct {
result *models.AnalysisResult
detectedVendor string
jobManager *JobManager
jobManager *JobManager
collectors *collector.Registry
}
func New(cfg Config) *Server {
@@ -37,6 +39,7 @@ func New(cfg Config) *Server {
config: cfg,
mux: http.NewServeMux(),
jobManager: NewJobManager(),
collectors: collector.NewDefaultRegistry(),
}
s.setupRoutes()
return s

9
internal/server/source_metadata_test.go
View File

@@ -30,7 +30,7 @@ func TestApplyArchiveSourceMetadata(t *testing.T) {
}
}
func TestNewAPIResultMetadata(t *testing.T) {
func TestApplyCollectSourceMetadata(t *testing.T) {
req := CollectRequest{
Host: "bmc-api.local",
Protocol: "redfish",
@@ -41,7 +41,12 @@ func TestNewAPIResultMetadata(t *testing.T) {
TLSMode: "strict",
}
result := newAPIResult(req)
result := &models.AnalysisResult{
Events: make([]models.Event, 0),
FRU: make([]models.FRUInfo, 0),
Sensors: make([]models.SensorReading, 0),
}
applyCollectSourceMetadata(result, req)
if result.SourceType != models.SourceTypeAPI {
t.Fatalf("expected source type %q, got %q", models.SourceTypeAPI, result.SourceType)

3
internal/server/upload_live_smoke_test.go
View File

@@ -15,7 +15,8 @@ import (
func newFlowTestServer() (*Server, *httptest.Server) {
s := &Server{
jobManager: NewJobManager(),
jobManager: NewJobManager(),
collectors: testCollectorRegistry(),
}
mux := http.NewServeMux()
mux.HandleFunc("POST /api/upload", s.handleUpload)