40b8afb150
Backend stability and security improvements: * internal/util/ — common RandomHex helper, removed 3 duplicates * ESL: deduplicated readMessage (locked/unlocked), net.JoinHostPort for IPv6 * AMI: synchronous reconnect() in readEventsLoop, net.JoinHostPort for IPv6 * Auth: /api/auth/refresh accepts Authorization header only (no ?token=) * decodeJSON: http.MaxBytesReader(1<<20) body limit * Trunks: gatewayParams() uses configured ESL.GatewayPrefix * Config: jwt_secret_env env-var fallback * FSCollector: time.After → time.NewTimer with defer Stop * Monitoring: Prometheus counters (route_requests, nodes_total/healthy, uptime) * go fmt pass across all internal/ packages
58 lines
1.3 KiB
Go
58 lines
1.3 KiB
Go
package agent
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
|
|
natsgo "github.com/nats-io/nats.go"
|
|
|
|
"github.com/pulse-lets-go/internal/models"
|
|
)
|
|
|
|
// Publisher публикует метрики в NATS.
|
|
type Publisher struct {
|
|
nc *natsgo.Conn
|
|
nodeID string
|
|
}
|
|
|
|
// NewPublisher создаёт NATS publisher.
|
|
func NewPublisher(natsURL, user, password, nodeID string) (*Publisher, error) {
|
|
opts := []natsgo.Option{
|
|
natsgo.ReconnectWait(natsgo.DefaultReconnectWait),
|
|
natsgo.MaxReconnects(-1),
|
|
}
|
|
if user != "" && password != "" {
|
|
opts = append(opts, natsgo.UserInfo(user, password))
|
|
}
|
|
|
|
nc, err := natsgo.Connect(natsURL, opts...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("nats connect: %w", err)
|
|
}
|
|
|
|
log.Printf("[publisher] NATS подключён к %s", natsURL)
|
|
return &Publisher{nc: nc, nodeID: nodeID}, nil
|
|
}
|
|
|
|
// Publish отправляет метрику в NATS: pulse.metrics.<node_id>.
|
|
func (p *Publisher) Publish(m *models.NodeMetric) error {
|
|
subject := fmt.Sprintf("pulse.metrics.%s", p.nodeID)
|
|
data, err := json.Marshal(m)
|
|
if err != nil {
|
|
return fmt.Errorf("marshal metric: %w", err)
|
|
}
|
|
|
|
if err := p.nc.Publish(subject, data); err != nil {
|
|
return fmt.Errorf("publish: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Close закрывает NATS-соединение.
|
|
func (p *Publisher) Close() {
|
|
if p.nc != nil {
|
|
p.nc.Close()
|
|
}
|
|
}
|