You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
280 lines
9.5 KiB
280 lines
9.5 KiB
// tera-seriald owns validated CH01-CH48 line configuration for TERA boards.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"net"
|
|
"os"
|
|
"os/signal"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"tera-clone/services/tera-api/internal/domain"
|
|
"tera-clone/services/tera-api/internal/logservice"
|
|
"tera-clone/services/tera-api/internal/logstorage"
|
|
"tera-clone/services/tera-api/internal/serial"
|
|
)
|
|
|
|
func main() {
|
|
controlSocketPath := getenv("TERA_SERIALD_SOCKET", serial.DefaultSocketPath)
|
|
streamSocketPath := getenv("TERA_SERIALD_STREAM_SOCKET", serial.DefaultStreamSocketPath)
|
|
mappingPath := getenv("TERA_SERIALD_MAPPING", "/etc/tera/serial-map.json")
|
|
hardwareProfile := getenv("TERA_SERIALD_HARDWARE_PROFILE", "fet3568j-c")
|
|
controller, err := serial.LoadMappingController(mappingPath, hardwareProfile)
|
|
if err != nil {
|
|
log.Printf("tera-seriald is fail-closed: %v", err)
|
|
controller = nil
|
|
}
|
|
if controller != nil {
|
|
defer func() {
|
|
if err := controller.Close(); err != nil {
|
|
log.Printf("tera-seriald serial hold close failed: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
controlListener, err := listen(controlSocketPath)
|
|
if err != nil {
|
|
log.Fatalf("tera-seriald control socket setup failed: %v", err)
|
|
}
|
|
streamListener, err := listen(streamSocketPath)
|
|
if err != nil {
|
|
_ = controlListener.Close()
|
|
_ = os.Remove(controlSocketPath)
|
|
log.Fatalf("tera-seriald stream socket setup failed: %v", err)
|
|
}
|
|
defer func() {
|
|
_ = controlListener.Close()
|
|
_ = streamListener.Close()
|
|
_ = os.Remove(controlSocketPath)
|
|
_ = os.Remove(streamSocketPath)
|
|
}()
|
|
|
|
unavailable := serial.UnavailableController{Reason: "validated serial mapping is not installed"}
|
|
var hardware serial.Controller = unavailable
|
|
var streamResolver serial.StreamResolver = unavailable
|
|
if controller != nil {
|
|
hardware = controller
|
|
streamResolver = controller
|
|
}
|
|
eventSink := serial.NewBoundedStreamEventSink(getenvInt("TERA_SERIALD_EVENT_QUEUE", 1024))
|
|
streamManager, err := serial.NewStreamLeaseManager(streamResolver, serial.StreamLeaseOptions{Events: eventSink})
|
|
if err != nil {
|
|
log.Fatalf("tera-seriald stream manager setup failed: %v", err)
|
|
}
|
|
reporterContext, stopReporter := context.WithCancel(context.Background())
|
|
defer stopReporter()
|
|
var linkMonitor serial.LinkStateController = serial.DisabledLinkStateController{
|
|
Reason: "CTS/DSR/DCD are excluded from unified RX/CR connection status",
|
|
}
|
|
if controller != nil && getenvBool("TERA_SERIALD_LINK_SCAN_ENABLED", false) {
|
|
controller.SetLinkWorkerCount(getenvInt("TERA_SERIALD_LINK_WORKERS", 8))
|
|
monitor, monitorErr := serial.NewLinkMonitor(controller, serial.LinkMonitorOptions{
|
|
Interval: time.Duration(getenvInt("TERA_SERIALD_LINK_SCAN_INTERVAL_MS", 1000)) * time.Millisecond,
|
|
})
|
|
if monitorErr != nil {
|
|
log.Fatalf("tera-seriald link monitor setup failed: %v", monitorErr)
|
|
}
|
|
linkMonitor = monitor
|
|
go monitor.Run(reporterContext)
|
|
}
|
|
crProbeEnabled := controller != nil && getenvBool("TERA_SERIALD_CR_PROBE_ENABLED", true)
|
|
var crProbeGroups []serial.CRProbeGroup
|
|
if controller != nil {
|
|
crProbeGroups = controller.CRProbeGroups()
|
|
}
|
|
crProbeManager, err := serial.NewCRProbeManager(streamManager, serial.CRProbeOptions{
|
|
Enabled: crProbeEnabled,
|
|
ResponseTimeout: time.Duration(getenvInt("TERA_SERIALD_CR_PROBE_TIMEOUT_MS", 100)) * time.Millisecond,
|
|
SweepInterval: time.Duration(getenvInt("TERA_SERIALD_CR_PROBE_SWEEP_SECONDS", 4)) * time.Second,
|
|
BetweenChannels: time.Duration(getenvInt("TERA_SERIALD_CR_PROBE_SPACING_MS", 20)) * time.Millisecond,
|
|
QuietPeriod: time.Duration(getenvInt("TERA_SERIALD_CR_PROBE_QUIET_SECONDS", 5)) * time.Second,
|
|
StartupDelay: time.Duration(getenvInt("TERA_SERIALD_CR_PROBE_STARTUP_DELAY_SECONDS", 30)) * time.Second,
|
|
Workers: getenvInt("TERA_SERIALD_CR_PROBE_WORKERS", 2),
|
|
Groups: crProbeGroups,
|
|
ConfirmResponses: getenvInt("TERA_SERIALD_CR_PROBE_CONFIRM_RESPONSES", 1),
|
|
DownMisses: getenvInt("TERA_SERIALD_CR_PROBE_DOWN_MISSES", 2),
|
|
})
|
|
if err != nil {
|
|
log.Fatalf("tera-seriald CR probe setup failed: %v", err)
|
|
}
|
|
if crProbeEnabled {
|
|
for _, group := range crProbeGroups {
|
|
log.Printf("tera-seriald CR probe group pci_bdf=%s channels=%d", group.ID, len(group.Channels))
|
|
}
|
|
}
|
|
go crProbeManager.Run(reporterContext)
|
|
logJournal := logservice.Journal(logservice.NewClient(getenv("TERA_SERIALD_LOG_SOCKET", logservice.DefaultSocketPath)))
|
|
go reportStreamEvents(reporterContext, streamManager, eventSink.Events(), logJournal, time.Duration(getenvInt("TERA_SERIALD_REPORT_INTERVAL_SECONDS", 10))*time.Second)
|
|
controlServer := serial.SocketServer{Controller: hardware, Capture: streamManager, CRProbe: crProbeManager, Links: linkMonitor}
|
|
streamServer := serial.StreamSocketServer{Manager: streamManager}
|
|
done := make(chan error, 2)
|
|
go func() { done <- controlServer.Serve(controlListener) }()
|
|
go func() { done <- streamServer.Serve(streamListener) }()
|
|
log.Printf("tera-seriald serving validated serial configuration on %s", controlSocketPath)
|
|
log.Printf("tera-seriald serving exclusive serial streams on %s", streamSocketPath)
|
|
|
|
signals := make(chan os.Signal, 1)
|
|
signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
|
|
select {
|
|
case signal := <-signals:
|
|
log.Printf("tera-seriald stopping after %s", signal)
|
|
stopReporter()
|
|
_ = controlListener.Close()
|
|
_ = streamListener.Close()
|
|
_ = streamManager.Close()
|
|
for range 2 {
|
|
if err := <-done; err != nil {
|
|
log.Printf("tera-seriald stop error: %v", err)
|
|
}
|
|
}
|
|
case err := <-done:
|
|
stopReporter()
|
|
_ = controlListener.Close()
|
|
_ = streamListener.Close()
|
|
_ = streamManager.Close()
|
|
if err != nil {
|
|
log.Fatalf("tera-seriald failed: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func listen(socketPath string) (net.Listener, error) {
|
|
if socketPath == "" || !filepath.IsAbs(socketPath) {
|
|
return nil, errors.New("TERA_SERIALD_SOCKET must be an absolute path")
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(socketPath), 0o750); err != nil {
|
|
return nil, err
|
|
}
|
|
if info, err := os.Lstat(socketPath); err == nil {
|
|
if info.Mode()&os.ModeSocket == 0 {
|
|
return nil, errors.New("refusing to replace non-socket serial control path")
|
|
}
|
|
if err := os.Remove(socketPath); err != nil {
|
|
return nil, err
|
|
}
|
|
} else if !os.IsNotExist(err) {
|
|
return nil, err
|
|
}
|
|
listener, err := net.Listen("unix", socketPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := os.Chmod(socketPath, 0o660); err != nil {
|
|
_ = listener.Close()
|
|
return nil, err
|
|
}
|
|
return listener, nil
|
|
}
|
|
|
|
func getenv(key, fallback string) string {
|
|
if value := os.Getenv(key); value != "" {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func getenvInt(key string, fallback int) int {
|
|
value := os.Getenv(key)
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
parsed, err := strconv.Atoi(value)
|
|
if err != nil || parsed <= 0 {
|
|
return fallback
|
|
}
|
|
return parsed
|
|
}
|
|
|
|
func getenvBool(key string, fallback bool) bool {
|
|
value := os.Getenv(key)
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
parsed, err := strconv.ParseBool(value)
|
|
if err != nil {
|
|
return fallback
|
|
}
|
|
return parsed
|
|
}
|
|
|
|
type streamIntervalTotal struct {
|
|
rx uint64
|
|
tx uint64
|
|
}
|
|
|
|
func reportStreamEvents(ctx context.Context, manager *serial.StreamLeaseManager, events <-chan serial.StreamEvent, journal logservice.Journal, interval time.Duration) {
|
|
if interval <= 0 {
|
|
interval = 10 * time.Second
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
totals := make(map[string]streamIntervalTotal)
|
|
var storageFailures uint64
|
|
var lastStorageError error
|
|
flush := func() {
|
|
for key, total := range totals {
|
|
if total.rx != 0 || total.tx != 0 {
|
|
log.Printf("tera-seriald stream interval lease=%s rx_bytes=%d tx_bytes=%d", key, total.rx, total.tx)
|
|
}
|
|
delete(totals, key)
|
|
}
|
|
snapshot := manager.Snapshot()
|
|
log.Printf("tera-seriald stream status active_leases=%d rx_bytes=%d tx_bytes=%d dropped_events=%d storage_failures=%d", snapshot.ActiveLeases, snapshot.RXBytes, snapshot.TXBytes, snapshot.DroppedEvents, storageFailures)
|
|
if lastStorageError != nil {
|
|
log.Printf("tera-seriald payload logging degraded: %v", lastStorageError)
|
|
lastStorageError = nil
|
|
}
|
|
}
|
|
defer flush()
|
|
for {
|
|
select {
|
|
case event := <-events:
|
|
key := event.Channel + "/" + event.LeaseID
|
|
switch event.Direction {
|
|
case "open", "close":
|
|
log.Printf("tera-seriald stream lifecycle channel=%s lease=%s client=%s action=%s", event.Channel, event.LeaseID, event.ClientID, event.Direction)
|
|
case "rx":
|
|
total := totals[key]
|
|
total.rx += event.Bytes
|
|
totals[key] = total
|
|
if err := recordPayload(journal, event); err != nil {
|
|
storageFailures++
|
|
lastStorageError = err
|
|
}
|
|
case "tx":
|
|
total := totals[key]
|
|
total.tx += event.Bytes
|
|
totals[key] = total
|
|
if err := recordPayload(journal, event); err != nil {
|
|
storageFailures++
|
|
lastStorageError = err
|
|
}
|
|
}
|
|
case <-ticker.C:
|
|
flush()
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func recordPayload(journal logservice.Journal, event serial.StreamEvent) error {
|
|
if journal == nil || len(event.Payload) == 0 {
|
|
return nil
|
|
}
|
|
record := domain.PortLogEvent{
|
|
Timestamp: event.Timestamp, Channel: event.Channel, EventType: "traffic",
|
|
Direction: strings.ToUpper(event.Direction), Actor: event.ClientID,
|
|
Message: fmt.Sprintf("%s %d serial payload byte(s)", strings.ToUpper(event.Direction), len(event.Payload)),
|
|
Payload: append([]byte(nil), event.Payload...), PayloadBytes: len(event.Payload), PayloadEncoding: "base64",
|
|
DisplayFormat: event.LogFormat, ShowTimestamp: event.ShowTimestamp, ShowDirection: event.ShowDirection,
|
|
ShowUsername: event.ShowUsername, ShowIPAddress: event.ShowIPAddress, ShowPort: event.ShowPort,
|
|
}
|
|
return journal.Append(logstorage.ClassPort, event.Timestamp, record)
|
|
}
|
|
|