Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 advertise.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "errors" "fmt" "net" "github.com/hashicorp/go-sockaddr" ) type getIPFunc func() (string, error) // These are overridden in unit tests to mock the sockaddr functions. var ( getPrivateAddress getIPFunc = sockaddr.GetPrivateIP getPublicAddress getIPFunc = sockaddr.GetPublicIP ) // calculateAdvertiseAddress attempts to clone logic from deep within memberlist // (NetTransport.FinalAdvertiseAddr) in order to surface its conclusions to the // application, so we can provide more actionable error messages if the user has // inadvertently misconfigured their cluster. // // https://github.com/hashicorp/memberlist/blob/022f081/net_transport.go#L126 func calculateAdvertiseAddress(bindAddr, advertiseAddr string, allowInsecureAdvertise bool) (net.IP, error) { if advertiseAddr != "" { ip := net.ParseIP(advertiseAddr) if ip == nil { return nil, fmt.Errorf("failed to parse advertise addr '%s'", advertiseAddr) } if ip4 := ip.To4(); ip4 != nil { ip = ip4 } return ip, nil } if isAny(bindAddr) { return discoverAdvertiseAddress(allowInsecureAdvertise) } ip := net.ParseIP(bindAddr) if ip == nil { return nil, fmt.Errorf("failed to parse bind addr '%s'", bindAddr) } return ip, nil } // discoverAdvertiseAddress will attempt to get a single IP address to use as // the advertise address when one is not explicitly provided. It defaults to // using a private IP address, and if not found then using a public IP if // insecure advertising is allowed. func discoverAdvertiseAddress(allowInsecureAdvertise bool) (net.IP, error) { addr, err := getPrivateAddress() if err != nil { return nil, fmt.Errorf("failed to get private IP: %w", err) } if addr == "" && !allowInsecureAdvertise { return nil, errors.New("no private IP found, explicit advertise addr not provided") } if addr == "" { addr, err = getPublicAddress() if err != nil { return nil, fmt.Errorf("failed to get public IP: %w", err) } if addr == "" { return nil, errors.New("no private/public IP found, explicit advertise addr not provided") } } ip := net.ParseIP(addr) if ip == nil { return nil, fmt.Errorf("failed to parse discovered IP '%s'", addr) } return ip, nil }	2.74 KB	2025-05-25 12:27
📄 advertise_test.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "errors" "net" "testing" "github.com/stretchr/testify/require" ) func TestCalculateAdvertiseAddress(t testing.T) { old := getPrivateAddress defer func() { getPrivateAddress = old }() cases := []struct { name string privateIPFn getIPFunc publicIPFn getIPFunc bind, advertise string allowInsecureAdvertise bool expectedIP net.IP err bool }{ { name: "use provided bind address", bind: "192.0.2.1", advertise: "", expectedIP: net.ParseIP("192.0.2.1"), err: false, }, { name: "use provided advertise address", bind: "192.0.2.1", advertise: "192.0.2.2", expectedIP: net.ParseIP("192.0.2.2"), err: false, }, { name: "discover private ip address", privateIPFn: func() (string, error) { return "192.0.2.1", nil }, bind: "0.0.0.0", advertise: "", expectedIP: net.ParseIP("192.0.2.1"), err: false, }, { name: "error if getPrivateAddress errors", privateIPFn: func() (string, error) { return "", errors.New("some error") }, bind: "0.0.0.0", advertise: "", err: true, }, { name: "error if getPrivateAddress returns an invalid address", privateIPFn: func() (string, error) { return "invalid", nil }, bind: "0.0.0.0", advertise: "", err: true, }, { name: "error if getPrivateAddress returns an empty address", privateIPFn: func() (string, error) { return "", nil }, bind: "0.0.0.0", advertise: "", err: true, }, { name: "discover public advertise address", privateIPFn: func() (string, error) { return "", nil }, publicIPFn: func() (string, error) { return "192.0.2.1", nil }, bind: "0.0.0.0", advertise: "", allowInsecureAdvertise: true, expectedIP: net.ParseIP("192.0.2.1"), err: false, }, { name: "error if getPublicAddress errors", privateIPFn: func() (string, error) { return "", nil }, publicIPFn: func() (string, error) { return "", errors.New("some error") }, bind: "0.0.0.0", advertise: "", allowInsecureAdvertise: true, err: true, }, { name: "error if getPublicAddress returns an invalid address", privateIPFn: func() (string, error) { return "", nil }, publicIPFn: func() (string, error) { return "invalid", nil }, bind: "0.0.0.0", advertise: "", allowInsecureAdvertise: true, err: true, }, { name: "error if getPublicAddress returns an empty address", privateIPFn: func() (string, error) { return "", nil }, publicIPFn: func() (string, error) { return "", nil }, bind: "0.0.0.0", advertise: "", allowInsecureAdvertise: true, err: true, }, } for _, c := range cases { t.Run(c.name, func(t testing.T) { getPrivateAddress = c.privateIPFn getPublicAddress = c.publicIPFn got, err := calculateAdvertiseAddress(c.bind, c.advertise, c.allowInsecureAdvertise) if c.err { require.Error(t, err) return } require.NoError(t, err) require.Equal(t, c.expectedIP.String(), got.String()) }) } }	3.95 KB	2025-05-25 12:27
📄 channel.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "log/slog" "sync" "time" "github.com/gogo/protobuf/proto" "github.com/hashicorp/memberlist" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/alertmanager/cluster/clusterpb" ) // Channel allows clients to send messages for a specific state type that will be // broadcasted in a best-effort manner. type Channel struct { key string send func([]byte) peers func() []memberlist.Node sendOversize func(memberlist.Node, []byte) error msgc chan []byte logger slog.Logger oversizeGossipMessageFailureTotal prometheus.Counter oversizeGossipMessageDroppedTotal prometheus.Counter oversizeGossipMessageSentTotal prometheus.Counter oversizeGossipDuration prometheus.Histogram } // NewChannel creates a new Channel struct, which handles sending normal and // oversize messages to peers. func NewChannel( key string, send func([]byte), peers func() []memberlist.Node, sendOversize func(memberlist.Node, []byte) error, logger slog.Logger, stopc chan struct{}, reg prometheus.Registerer, ) Channel { oversizeGossipMessageFailureTotal := prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_oversized_gossip_message_failure_total", Help: "Number of oversized gossip message sends that failed.", ConstLabels: prometheus.Labels{"key": key}, }) oversizeGossipMessageSentTotal := prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_oversized_gossip_message_sent_total", Help: "Number of oversized gossip message sent.", ConstLabels: prometheus.Labels{"key": key}, }) oversizeGossipMessageDroppedTotal := prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_oversized_gossip_message_dropped_total", Help: "Number of oversized gossip messages that were dropped due to a full message queue.", ConstLabels: prometheus.Labels{"key": key}, }) oversizeGossipDuration := prometheus.NewHistogram(prometheus.HistogramOpts{ Name: "alertmanager_oversize_gossip_message_duration_seconds", Help: "Duration of oversized gossip message requests.", ConstLabels: prometheus.Labels{"key": key}, Buckets: prometheus.DefBuckets, NativeHistogramBucketFactor: 1.1, NativeHistogramMaxBucketNumber: 100, NativeHistogramMinResetDuration: 1 time.Hour, }) reg.MustRegister(oversizeGossipDuration, oversizeGossipMessageFailureTotal, oversizeGossipMessageDroppedTotal, oversizeGossipMessageSentTotal) c := &Channel{ key: key, send: send, peers: peers, logger: logger, msgc: make(chan []byte, 200), sendOversize: sendOversize, oversizeGossipMessageFailureTotal: oversizeGossipMessageFailureTotal, oversizeGossipMessageDroppedTotal: oversizeGossipMessageDroppedTotal, oversizeGossipMessageSentTotal: oversizeGossipMessageSentTotal, oversizeGossipDuration: oversizeGossipDuration, } go c.handleOverSizedMessages(stopc) return c } // handleOverSizedMessages prevents memberlist from opening too many parallel // TCP connections to its peers. func (c Channel) handleOverSizedMessages(stopc chan struct{}) { var wg sync.WaitGroup for { select { case b := <-c.msgc: for _, n := range c.peers() { wg.Add(1) go func(n memberlist.Node) { defer wg.Done() c.oversizeGossipMessageSentTotal.Inc() start := time.Now() if err := c.sendOversize(n, b); err != nil { c.logger.Debug("failed to send reliable", "key", c.key, "node", n, "err", err) c.oversizeGossipMessageFailureTotal.Inc() return } c.oversizeGossipDuration.Observe(time.Since(start).Seconds()) }(n) } wg.Wait() case <-stopc: return } } } // Broadcast enqueues a message for broadcasting. func (c *Channel) Broadcast(b []byte) { b, err := proto.Marshal(&clusterpb.Part{Key: c.key, Data: b}) if err != nil { return } if OversizedMessage(b) { select { case c.msgc <- b: default: c.logger.Debug("oversized gossip channel full") c.oversizeGossipMessageDroppedTotal.Inc() } } else { c.send(b) } } // OversizedMessage indicates whether or not the byte payload should be sent // via TCP. func OversizedMessage(b []byte) bool { return len(b) > MaxGossipPacketSize/2 }	5 KB	2025-05-25 12:27
📄 channel_test.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "bytes" "context" "io" "os" "testing" "github.com/hashicorp/memberlist" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/promslog" ) func TestNormalMessagesGossiped(t testing.T) { var sent bool c := newChannel( func(_ []byte) { sent = true }, func() []memberlist.Node { return nil }, func(_ memberlist.Node, _ []byte) error { return nil }, ) c.Broadcast([]byte{}) if sent != true { t.Fatalf("small message not sent") } } func TestOversizedMessagesGossiped(t testing.T) { var sent bool ctx, cancel := context.WithCancel(context.Background()) c := newChannel( func(_ []byte) {}, func() []memberlist.Node { return []memberlist.Node{{}} }, func(_ memberlist.Node, _ []byte) error { sent = true; cancel(); return nil }, ) f, err := os.Open("/dev/zero") if err != nil { t.Fatalf("failed to open /dev/zero: %v", err) } defer f.Close() buf := new(bytes.Buffer) toCopy := int64(800) if n, err := io.CopyN(buf, f, toCopy); err != nil { t.Fatalf("failed to copy bytes: %v", err) } else if n != toCopy { t.Fatalf("wanted to copy %d bytes, only copied %d", toCopy, n) } c.Broadcast(buf.Bytes()) <-ctx.Done() if sent != true { t.Fatalf("oversized message not sent") } } func newChannel( send func([]byte), peers func() []memberlist.Node, sendOversize func(memberlist.Node, []byte) error, ) Channel { return NewChannel( "test", send, peers, sendOversize, promslog.NewNopLogger(), make(chan struct{}), prometheus.NewRegistry(), ) }	2.1 KB	2025-05-25 12:27
📄 cluster.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "context" "errors" "fmt" "log/slog" "math/rand" "net" "sort" "strconv" "strings" "sync" "time" "github.com/hashicorp/memberlist" "github.com/oklog/ulid" "github.com/prometheus/client_golang/prometheus" ) // ClusterPeer represents a single Peer in a gossip cluster. type ClusterPeer interface { // Name returns the unique identifier of this peer in the cluster. Name() string // Status returns a status string representing the peer state. Status() string // Peers returns the peer nodes in the cluster. Peers() []ClusterMember } // ClusterMember interface that represents node peers in a cluster. type ClusterMember interface { // Name returns the name of the node Name() string // Address returns the IP address of the node Address() string } // ClusterChannel supports state broadcasting across peers. type ClusterChannel interface { Broadcast([]byte) } // Peer is a single peer in a gossip cluster. type Peer struct { mlist memberlist.Memberlist delegate delegate resolvedPeers []string mtx sync.RWMutex states map[string]State stopc chan struct{} readyc chan struct{} peerLock sync.RWMutex peers map[string]peer failedPeers []peer knownPeers []string advertiseAddr string failedReconnectionsCounter prometheus.Counter reconnectionsCounter prometheus.Counter failedRefreshCounter prometheus.Counter refreshCounter prometheus.Counter peerLeaveCounter prometheus.Counter peerUpdateCounter prometheus.Counter peerJoinCounter prometheus.Counter logger slog.Logger } // peer is an internal type used for bookkeeping. It holds the state of peers // in the cluster. type peer struct { status PeerStatus leaveTime time.Time memberlist.Node } // PeerStatus is the state that a peer is in. type PeerStatus int const ( StatusNone PeerStatus = iota StatusAlive StatusFailed ) func (s PeerStatus) String() string { switch s { case StatusNone: return "none" case StatusAlive: return "alive" case StatusFailed: return "failed" default: panic(fmt.Sprintf("unknown PeerStatus: %d", s)) } } const ( DefaultPushPullInterval = 60 * time.Second DefaultGossipInterval = 200 * time.Millisecond DefaultTCPTimeout = 10 * time.Second DefaultProbeTimeout = 500 * time.Millisecond DefaultProbeInterval = 1 * time.Second DefaultReconnectInterval = 10 * time.Second DefaultReconnectTimeout = 6 * time.Hour DefaultRefreshInterval = 15 * time.Second MaxGossipPacketSize = 1400 ) func Create( l slog.Logger, reg prometheus.Registerer, bindAddr string, advertiseAddr string, knownPeers []string, waitIfEmpty bool, pushPullInterval time.Duration, gossipInterval time.Duration, tcpTimeout time.Duration, probeTimeout time.Duration, probeInterval time.Duration, tlsTransportConfig TLSTransportConfig, allowInsecureAdvertise bool, label string, ) (Peer, error) { bindHost, bindPortStr, err := net.SplitHostPort(bindAddr) if err != nil { return nil, fmt.Errorf("invalid listen address: %w", err) } bindPort, err := strconv.Atoi(bindPortStr) if err != nil { return nil, fmt.Errorf("address %s: invalid port: %w", bindAddr, err) } var advertiseHost string var advertisePort int if advertiseAddr != "" { var advertisePortStr string advertiseHost, advertisePortStr, err = net.SplitHostPort(advertiseAddr) if err != nil { return nil, fmt.Errorf("invalid advertise address: %w", err) } advertisePort, err = strconv.Atoi(advertisePortStr) if err != nil { return nil, fmt.Errorf("address %s: invalid port: %w", advertiseAddr, err) } } resolvedPeers, err := resolvePeers(context.Background(), knownPeers, advertiseAddr, &net.Resolver{}, waitIfEmpty) if err != nil { return nil, fmt.Errorf("resolve peers: %w", err) } l.Debug("resolved peers to following addresses", "peers", strings.Join(resolvedPeers, ",")) // Initial validation of user-specified advertise address. addr, err := calculateAdvertiseAddress(bindHost, advertiseHost, allowInsecureAdvertise) if err != nil { l.Warn("couldn't deduce an advertise address: " + err.Error()) } else if hasNonlocal(resolvedPeers) && isUnroutable(addr.String()) { l.Warn("this node advertises itself on an unroutable address", "addr", addr.String()) l.Warn("this node will be unreachable in the cluster") l.Warn("provide --cluster.advertise-address as a routable IP address or hostname") } else if isAny(bindAddr) && advertiseHost == "" { // memberlist doesn't advertise properly when the bind address is empty or unspecified. l.Info("setting advertise address explicitly", "addr", addr.String(), "port", bindPort) advertiseHost = addr.String() advertisePort = bindPort } // TODO(fabxc): generate human-readable but random names? name, err := ulid.New(ulid.Now(), rand.New(rand.NewSource(time.Now().UnixNano()))) if err != nil { return nil, err } p := &Peer{ states: map[string]State{}, stopc: make(chan struct{}), readyc: make(chan struct{}), logger: l, peers: map[string]peer{}, resolvedPeers: resolvedPeers, knownPeers: knownPeers, } p.register(reg, name.String()) retransmit := len(knownPeers) / 2 if retransmit < 3 { retransmit = 3 } p.delegate = newDelegate(l, reg, p, retransmit) cfg := memberlist.DefaultLANConfig() cfg.Name = name.String() cfg.BindAddr = bindHost cfg.BindPort = bindPort cfg.Delegate = p.delegate cfg.Ping = p.delegate cfg.Alive = p.delegate cfg.Events = p.delegate cfg.GossipInterval = gossipInterval cfg.PushPullInterval = pushPullInterval cfg.TCPTimeout = tcpTimeout cfg.ProbeTimeout = probeTimeout cfg.ProbeInterval = probeInterval cfg.Logger = slog.NewLogLogger(l.Handler(), slog.LevelDebug) cfg.GossipNodes = retransmit cfg.UDPBufferSize = MaxGossipPacketSize cfg.Label = label if advertiseHost != "" { cfg.AdvertiseAddr = advertiseHost cfg.AdvertisePort = advertisePort p.setInitialFailed(resolvedPeers, fmt.Sprintf("%s:%d", advertiseHost, advertisePort)) } else { p.setInitialFailed(resolvedPeers, bindAddr) } if tlsTransportConfig != nil { l.Info("using TLS for gossip") cfg.Transport, err = NewTLSTransport(context.Background(), l, reg, cfg.BindAddr, cfg.BindPort, tlsTransportConfig) if err != nil { return nil, fmt.Errorf("tls transport: %w", err) } } ml, err := memberlist.Create(cfg) if err != nil { return nil, fmt.Errorf("create memberlist: %w", err) } p.mlist = ml return p, nil } func (p Peer) Join( reconnectInterval time.Duration, reconnectTimeout time.Duration, ) error { n, err := p.mlist.Join(p.resolvedPeers) if err != nil { p.logger.Warn("failed to join cluster", "err", err) if reconnectInterval != 0 { p.logger.Info(fmt.Sprintf("will retry joining cluster every %v", reconnectInterval.String())) } } else { p.logger.Debug("joined cluster", "peers", n) } if reconnectInterval != 0 { go p.runPeriodicTask( reconnectInterval, p.reconnect, ) } if reconnectTimeout != 0 { go p.runPeriodicTask( 5time.Minute, func() { p.removeFailedPeers(reconnectTimeout) }, ) } go p.runPeriodicTask( DefaultRefreshInterval, p.refresh, ) return err } // All peers are initially added to the failed list. They will be removed from // this list in peerJoin when making their initial connection. func (p Peer) setInitialFailed(peers []string, myAddr string) { if len(peers) == 0 { return } p.peerLock.Lock() defer p.peerLock.Unlock() now := time.Now() for _, peerAddr := range peers { if peerAddr == myAddr { // Don't add ourselves to the initially failing list, // we don't connect to ourselves. continue } host, port, err := net.SplitHostPort(peerAddr) if err != nil { continue } ip := net.ParseIP(host) if ip == nil { // Don't add textual addresses since memberlist only advertises // dotted decimal or IPv6 addresses. continue } portUint, err := strconv.ParseUint(port, 10, 16) if err != nil { continue } pr := peer{ status: StatusFailed, leaveTime: now, Node: &memberlist.Node{ Addr: ip, Port: uint16(portUint), }, } p.failedPeers = append(p.failedPeers, pr) p.peers[peerAddr] = pr } } func (p Peer) register(reg prometheus.Registerer, name string) { peerInfo := prometheus.NewGauge( prometheus.GaugeOpts{ Name: "alertmanager_cluster_peer_info", Help: "A metric with a constant '1' value labeled by peer name.", ConstLabels: prometheus.Labels{"peer": name}, }, ) peerInfo.Set(1) clusterFailedPeers := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "alertmanager_cluster_failed_peers", Help: "Number indicating the current number of failed peers in the cluster.", }, func() float64 { p.peerLock.RLock() defer p.peerLock.RUnlock() return float64(len(p.failedPeers)) }) p.failedReconnectionsCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_reconnections_failed_total", Help: "A counter of the number of failed cluster peer reconnection attempts.", }) p.reconnectionsCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_reconnections_total", Help: "A counter of the number of cluster peer reconnections.", }) p.failedRefreshCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_refresh_join_failed_total", Help: "A counter of the number of failed cluster peer joined attempts via refresh.", }) p.refreshCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_refresh_join_total", Help: "A counter of the number of cluster peer joined via refresh.", }) p.peerLeaveCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_peers_left_total", Help: "A counter of the number of peers that have left.", }) p.peerUpdateCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_peers_update_total", Help: "A counter of the number of peers that have updated metadata.", }) p.peerJoinCounter = prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_peers_joined_total", Help: "A counter of the number of peers that have joined.", }) reg.MustRegister(peerInfo, clusterFailedPeers, p.failedReconnectionsCounter, p.reconnectionsCounter, p.peerLeaveCounter, p.peerUpdateCounter, p.peerJoinCounter, p.refreshCounter, p.failedRefreshCounter) } func (p Peer) runPeriodicTask(d time.Duration, f func()) { tick := time.NewTicker(d) defer tick.Stop() for { select { case <-p.stopc: return case <-tick.C: f() } } } func (p Peer) removeFailedPeers(timeout time.Duration) { p.peerLock.Lock() defer p.peerLock.Unlock() now := time.Now() keep := make([]peer, 0, len(p.failedPeers)) for _, pr := range p.failedPeers { if pr.leaveTime.Add(timeout).After(now) { keep = append(keep, pr) } else { p.logger.Debug("failed peer has timed out", "peer", pr.Node, "addr", pr.Address()) delete(p.peers, pr.Name) } } p.failedPeers = keep } func (p Peer) reconnect() { p.peerLock.RLock() failedPeers := p.failedPeers p.peerLock.RUnlock() logger := p.logger.With("msg", "reconnect") for _, pr := range failedPeers { // No need to do book keeping on failedPeers here. If a // reconnect is successful, they will be announced in // peerJoin(). if _, err := p.mlist.Join([]string{pr.Address()}); err != nil { p.failedReconnectionsCounter.Inc() logger.Debug("failure", "peer", pr.Node, "addr", pr.Address(), "err", err) } else { p.reconnectionsCounter.Inc() logger.Debug("success", "peer", pr.Node, "addr", pr.Address()) } } } func (p Peer) refresh() { logger := p.logger.With("msg", "refresh") resolvedPeers, err := resolvePeers(context.Background(), p.knownPeers, p.advertiseAddr, &net.Resolver{}, false) if err != nil { logger.Debug(fmt.Sprintf("%v", p.knownPeers), "err", err) return } members := p.mlist.Members() for _, peer := range resolvedPeers { var isPeerFound bool for _, member := range members { if member.Address() == peer { isPeerFound = true break } } if !isPeerFound { if _, err := p.mlist.Join([]string{peer}); err != nil { p.failedRefreshCounter.Inc() logger.Warn("failure", "addr", peer, "err", err) } else { p.refreshCounter.Inc() logger.Debug("success", "addr", peer) } } } } func (p Peer) peerJoin(n memberlist.Node) { p.peerLock.Lock() defer p.peerLock.Unlock() var oldStatus PeerStatus pr, ok := p.peers[n.Address()] if !ok { oldStatus = StatusNone pr = peer{ status: StatusAlive, Node: n, } } else { oldStatus = pr.status pr.Node = n pr.status = StatusAlive pr.leaveTime = time.Time{} } p.peers[n.Address()] = pr p.peerJoinCounter.Inc() if oldStatus == StatusFailed { p.logger.Debug("peer rejoined", "peer", pr.Node) p.failedPeers = removeOldPeer(p.failedPeers, pr.Address()) } } func (p Peer) peerLeave(n memberlist.Node) { p.peerLock.Lock() defer p.peerLock.Unlock() pr, ok := p.peers[n.Address()] if !ok { // Why are we receiving a leave notification from a node that // never joined? return } pr.status = StatusFailed pr.leaveTime = time.Now() p.failedPeers = append(p.failedPeers, pr) p.peers[n.Address()] = pr p.peerLeaveCounter.Inc() p.logger.Debug("peer left", "peer", pr.Node) } func (p Peer) peerUpdate(n memberlist.Node) { p.peerLock.Lock() defer p.peerLock.Unlock() pr, ok := p.peers[n.Address()] if !ok { // Why are we receiving an update from a node that never // joined? return } pr.Node = n p.peers[n.Address()] = pr p.peerUpdateCounter.Inc() p.logger.Debug("peer updated", "peer", pr.Node) } // AddState adds a new state that will be gossiped. It returns a channel to which // broadcast messages for the state can be sent. func (p Peer) AddState(key string, s State, reg prometheus.Registerer) ClusterChannel { p.mtx.Lock() p.states[key] = s p.mtx.Unlock() send := func(b []byte) { p.delegate.bcast.QueueBroadcast(simpleBroadcast(b)) } peers := func() []memberlist.Node { nodes := p.mlist.Members() for i, n := range nodes { if n.String() == p.Self().Name { nodes = append(nodes[:i], nodes[i+1:]...) break } } return nodes } sendOversize := func(n memberlist.Node, b []byte) error { return p.mlist.SendReliable(n, b) } return NewChannel(key, send, peers, sendOversize, p.logger, p.stopc, reg) } // Leave the cluster, waiting up to timeout. func (p Peer) Leave(timeout time.Duration) error { close(p.stopc) p.logger.Debug("leaving cluster") return p.mlist.Leave(timeout) } // Name returns the unique ID of this peer in the cluster. func (p Peer) Name() string { return p.mlist.LocalNode().Name } // ClusterSize returns the current number of alive members in the cluster. func (p Peer) ClusterSize() int { return p.mlist.NumMembers() } // Return true when router has settled. func (p Peer) Ready() bool { select { case <-p.readyc: return true default: } return false } // Wait until Settle() has finished. func (p Peer) WaitReady(ctx context.Context) error { select { case <-ctx.Done(): return ctx.Err() case <-p.readyc: return nil } } // Return a status string representing the peer state. func (p Peer) Status() string { if p.Ready() { return "ready" } return "settling" } // Info returns a JSON-serializable dump of cluster state. // Useful for debug. func (p Peer) Info() map[string]interface{} { p.mtx.RLock() defer p.mtx.RUnlock() return map[string]interface{}{ "self": p.mlist.LocalNode(), "members": p.mlist.Members(), } } // Self returns the node information about the peer itself. func (p Peer) Self() memberlist.Node { return p.mlist.LocalNode() } // Member represents a member in the cluster. type Member struct { node memberlist.Node } // Name implements cluster.ClusterMember. func (m Member) Name() string { return m.node.Name } // Address implements cluster.ClusterMember. func (m Member) Address() string { return m.node.Address() } // Peers returns the peers in the cluster. func (p Peer) Peers() []ClusterMember { peers := make([]ClusterMember, 0, len(p.mlist.Members())) for _, member := range p.mlist.Members() { peers = append(peers, Member{ node: member, }) } return peers } // Position returns the position of the peer in the cluster. func (p Peer) Position() int { all := p.mlist.Members() sort.Slice(all, func(i, j int) bool { return all[i].Name < all[j].Name }) k := 0 for _, n := range all { if n.Name == p.Self().Name { break } k++ } return k } // Settle waits until the mesh is ready (and sets the appropriate internal state when it is). // The idea is that we don't want to start "working" before we get a chance to know most of the alerts and/or silences. // Inspired from https://github.com/apache/cassandra/blob/7a40abb6a5108688fb1b10c375bb751cbb782ea4/src/java/org/apache/cassandra/gms/Gossiper.java // This is clearly not perfect or strictly correct but should prevent the alertmanager to send notification before it is obviously not ready. // This is especially important for those that do not have persistent storage. func (p Peer) Settle(ctx context.Context, interval time.Duration) { const NumOkayRequired = 3 p.logger.Info("Waiting for gossip to settle...", "interval", interval) start := time.Now() nPeers := 0 nOkay := 0 totalPolls := 0 for { select { case <-ctx.Done(): elapsed := time.Since(start) p.logger.Info("gossip not settled but continuing anyway", "polls", totalPolls, "elapsed", elapsed) close(p.readyc) return case <-time.After(interval): } elapsed := time.Since(start) n := len(p.Peers()) if nOkay >= NumOkayRequired { p.logger.Info("gossip settled; proceeding", "elapsed", elapsed) break } if n == nPeers { nOkay++ p.logger.Debug("gossip looks settled", "elapsed", elapsed) } else { nOkay = 0 p.logger.Info("gossip not settled", "polls", totalPolls, "before", nPeers, "now", n, "elapsed", elapsed) } nPeers = n totalPolls++ } close(p.readyc) } // State is a piece of state that can be serialized and merged with other // serialized state. type State interface { // MarshalBinary serializes the underlying state. MarshalBinary() ([]byte, error) // Merge merges serialized state into the underlying state. Merge(b []byte) error } // We use a simple broadcast implementation in which items are never invalidated by others. type simpleBroadcast []byte func (b simpleBroadcast) Message() []byte { return []byte(b) } func (b simpleBroadcast) Invalidates(memberlist.Broadcast) bool { return false } func (b simpleBroadcast) Finished() {} func resolvePeers(ctx context.Context, peers []string, myAddress string, res net.Resolver, waitIfEmpty bool) ([]string, error) { var resolvedPeers []string for _, peer := range peers { host, port, err := net.SplitHostPort(peer) if err != nil { return nil, fmt.Errorf("split host/port for peer %s: %w", peer, err) } retryCtx, cancel := context.WithCancel(ctx) defer cancel() ips, err := res.LookupIPAddr(ctx, host) if err != nil { // Assume direct address. resolvedPeers = append(resolvedPeers, peer) continue } if len(ips) == 0 { var lookupErrSpotted bool err := retry(2*time.Second, retryCtx.Done(), func() error { if lookupErrSpotted { // We need to invoke cancel in next run of retry when lookupErrSpotted to preserve LookupIPAddr error. cancel() } ips, err = res.LookupIPAddr(retryCtx, host) if err != nil { lookupErrSpotted = true return fmt.Errorf("IP Addr lookup for peer %s: %w", peer, err) } ips = removeMyAddr(ips, port, myAddress) if len(ips) == 0 { if !waitIfEmpty { return nil } return errors.New("empty IPAddr result. Retrying") } return nil }) if err != nil { return nil, err } } for _, ip := range ips { resolvedPeers = append(resolvedPeers, net.JoinHostPort(ip.String(), port)) } } return resolvedPeers, nil } func removeMyAddr(ips []net.IPAddr, targetPort, myAddr string) []net.IPAddr { var result []net.IPAddr for _, ip := range ips { if net.JoinHostPort(ip.String(), targetPort) == myAddr { continue } result = append(result, ip) } return result } func hasNonlocal(clusterPeers []string) bool { for _, peer := range clusterPeers { if host, _, err := net.SplitHostPort(peer); err == nil { peer = host } if ip := net.ParseIP(peer); ip != nil && !ip.IsLoopback() { return true } else if ip == nil && strings.ToLower(peer) != "localhost" { return true } } return false } func isUnroutable(addr string) bool { if host, _, err := net.SplitHostPort(addr); err == nil { addr = host } if ip := net.ParseIP(addr); ip != nil && (ip.IsUnspecified() \|\| ip.IsLoopback()) { return true // typically 0.0.0.0 or localhost } else if ip == nil && strings.ToLower(addr) == "localhost" { return true } return false } func isAny(addr string) bool { if host, _, err := net.SplitHostPort(addr); err == nil { addr = host } return addr == "" \|\| net.ParseIP(addr).IsUnspecified() } // retry executes f every interval seconds until timeout or no error is returned from f. func retry(interval time.Duration, stopc <-chan struct{}, f func() error) error { tick := time.NewTicker(interval) defer tick.Stop() var err error for { if err = f(); err == nil { return nil } select { case <-stopc: return err case <-tick.C: } } } func removeOldPeer(old []peer, addr string) []peer { new := make([]peer, 0, len(old)) for _, p := range old { if p.Address() != addr { new = append(new, p) } } return new }	21.97 KB	2025-05-25 12:27
📄 cluster_test.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "context" "testing" "time" "github.com/hashicorp/go-sockaddr" "github.com/stretchr/testify/require" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/promslog" ) func TestClusterJoinAndReconnect(t testing.T) { ip, _ := sockaddr.GetPrivateIP() if ip == "" { t.Skipf("skipping tests because no private IP address can be found") return } t.Run("TestJoinLeave", testJoinLeave) t.Run("TestReconnect", testReconnect) t.Run("TestRemoveFailedPeers", testRemoveFailedPeers) t.Run("TestInitiallyFailingPeers", testInitiallyFailingPeers) t.Run("TestTLSConnection", testTLSConnection) } func testJoinLeave(t testing.T) { logger := promslog.NewNopLogger() p, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, nil, false, "", ) require.NoError(t, err) require.NotNil(t, p) err = p.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) require.False(t, p.Ready()) { ctx, cancel := context.WithCancel(context.Background()) cancel() require.Equal(t, context.Canceled, p.WaitReady(ctx)) } require.Equal(t, "settling", p.Status()) go p.Settle(context.Background(), 0time.Second) require.NoError(t, p.WaitReady(context.Background())) require.Equal(t, "ready", p.Status()) // Create the peer who joins the first. p2, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{p.Self().Address()}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, nil, false, "", ) require.NoError(t, err) require.NotNil(t, p2) err = p2.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) go p2.Settle(context.Background(), 0time.Second) require.NoError(t, p2.WaitReady(context.Background())) require.Equal(t, 2, p.ClusterSize()) p2.Leave(0 * time.Second) require.Equal(t, 1, p.ClusterSize()) require.Len(t, p.failedPeers, 1) require.Equal(t, p2.Self().Address(), p.peers[p2.Self().Address()].Node.Address()) require.Equal(t, p2.Name(), p.failedPeers[0].Name) } func testReconnect(t testing.T) { logger := promslog.NewNopLogger() p, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, nil, false, "", ) require.NoError(t, err) require.NotNil(t, p) err = p.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) go p.Settle(context.Background(), 0time.Second) require.NoError(t, p.WaitReady(context.Background())) p2, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, nil, false, "", ) require.NoError(t, err) require.NotNil(t, p2) err = p2.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) go p2.Settle(context.Background(), 0time.Second) require.NoError(t, p2.WaitReady(context.Background())) p.peerJoin(p2.Self()) p.peerLeave(p2.Self()) require.Equal(t, 1, p.ClusterSize()) require.Len(t, p.failedPeers, 1) p.reconnect() require.Equal(t, 2, p.ClusterSize()) require.Empty(t, p.failedPeers) require.Equal(t, StatusAlive, p.peers[p2.Self().Address()].status) } func testRemoveFailedPeers(t testing.T) { logger := promslog.NewNopLogger() p, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, nil, false, "", ) require.NoError(t, err) require.NotNil(t, p) err = p.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) n := p.Self() now := time.Now() p1 := peer{ status: StatusFailed, leaveTime: now, Node: n, } p2 := peer{ status: StatusFailed, leaveTime: now.Add(-time.Hour), Node: n, } p3 := peer{ status: StatusFailed, leaveTime: now.Add(30 * -time.Minute), Node: n, } p.failedPeers = []peer{p1, p2, p3} p.removeFailedPeers(30 * time.Minute) require.Len(t, p.failedPeers, 1) require.Equal(t, p1, p.failedPeers[0]) } func testInitiallyFailingPeers(t testing.T) { logger := promslog.NewNopLogger() myAddr := "1.2.3.4:5000" peerAddrs := []string{myAddr, "2.3.4.5:5000", "3.4.5.6:5000", "foo.example.com:5000"} p, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, nil, false, "", ) require.NoError(t, err) require.NotNil(t, p) err = p.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) p.setInitialFailed(peerAddrs, myAddr) // We shouldn't have added "our" bind addr and the FQDN address to the // failed peers list. require.Len(t, p.failedPeers, len(peerAddrs)-2) for _, addr := range peerAddrs { if addr == myAddr \|\| addr == "foo.example.com:5000" { continue } pr, ok := p.peers[addr] require.True(t, ok) require.Equal(t, StatusFailed.String(), pr.status.String()) require.Equal(t, addr, pr.Address()) expectedLen := len(p.failedPeers) - 1 p.peerJoin(pr.Node) require.Len(t, p.failedPeers, expectedLen) } } func testTLSConnection(t testing.T) { logger := promslog.NewNopLogger() tlsTransportConfig1, err := GetTLSTransportConfig("./testdata/tls_config_node1.yml") require.NoError(t, err) p1, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, tlsTransportConfig1, false, "", ) require.NoError(t, err) require.NotNil(t, p1) err = p1.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) require.False(t, p1.Ready()) require.Equal(t, "settling", p1.Status()) go p1.Settle(context.Background(), 0time.Second) p1.WaitReady(context.Background()) require.Equal(t, "ready", p1.Status()) // Create the peer who joins the first. tlsTransportConfig2, err := GetTLSTransportConfig("./testdata/tls_config_node2.yml") require.NoError(t, err) p2, err := Create( logger, prometheus.NewRegistry(), "127.0.0.1:0", "", []string{p1.Self().Address()}, true, DefaultPushPullInterval, DefaultGossipInterval, DefaultTCPTimeout, DefaultProbeTimeout, DefaultProbeInterval, tlsTransportConfig2, false, "", ) require.NoError(t, err) require.NotNil(t, p2) err = p2.Join( DefaultReconnectInterval, DefaultReconnectTimeout, ) require.NoError(t, err) go p2.Settle(context.Background(), 0time.Second) p2.WaitReady(context.Background()) require.Equal(t, "ready", p2.Status()) require.Equal(t, 2, p1.ClusterSize()) p2.Leave(0 * time.Second) require.Equal(t, 1, p1.ClusterSize()) require.Len(t, p1.failedPeers, 1) require.Equal(t, p2.Self().Address(), p1.peers[p2.Self().Address()].Node.Address()) require.Equal(t, p2.Name(), p1.failedPeers[0].Name) }	7.9 KB	2025-05-25 12:27
📁 clusterpb	-	2025-05-25 12:27
📄 connection_pool.go // Copyright 2020 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "crypto/tls" "errors" "fmt" "sync" "time" lru "github.com/hashicorp/golang-lru/v2" ) const capacity = 1024 type connectionPool struct { mtx sync.Mutex cache lru.Cache[string, tlsConn] tlsConfig tls.Config } func newConnectionPool(tlsClientCfg tls.Config) (connectionPool, error) { cache, err := lru.NewWithEvict( capacity, func(_ string, conn tlsConn) { conn.Close() }, ) if err != nil { return nil, fmt.Errorf("failed to create new LRU: %w", err) } return &connectionPool{ cache: cache, tlsConfig: tlsClientCfg, }, nil } // borrowConnection returns a tlsConn from the pool. The connection does not // need to be returned to the pool because each connection has its own locking. func (pool connectionPool) borrowConnection(addr string, timeout time.Duration) (tlsConn, error) { pool.mtx.Lock() defer pool.mtx.Unlock() if pool.cache == nil { return nil, errors.New("connection pool closed") } key := fmt.Sprintf("%s/%d", addr, int64(timeout)) conn, exists := pool.cache.Get(key) if exists && conn.alive() { return conn, nil } conn, err := dialTLSConn(addr, timeout, pool.tlsConfig) if err != nil { return nil, err } pool.cache.Add(key, conn) return conn, nil } func (pool connectionPool) shutdown() { pool.mtx.Lock() defer pool.mtx.Unlock() if pool.cache == nil { return } pool.cache.Purge() pool.cache = nil }	1.96 KB	2025-05-25 12:27
📄 delegate.go // Copyright 2018 Prometheus Team // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "log/slog" "time" "github.com/gogo/protobuf/proto" "github.com/hashicorp/memberlist" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/alertmanager/cluster/clusterpb" ) const ( // Maximum number of messages to be held in the queue. maxQueueSize = 4096 fullState = "full_state" update = "update" ) // delegate implements memberlist.Delegate and memberlist.EventDelegate // and broadcasts its peer's state in the cluster. type delegate struct { Peer logger slog.Logger bcast memberlist.TransmitLimitedQueue messagesReceived prometheus.CounterVec messagesReceivedSize prometheus.CounterVec messagesSent prometheus.CounterVec messagesSentSize prometheus.CounterVec messagesPruned prometheus.Counter nodeAlive prometheus.CounterVec nodePingDuration prometheus.HistogramVec } func newDelegate(l slog.Logger, reg prometheus.Registerer, p Peer, retransmit int) delegate { bcast := &memberlist.TransmitLimitedQueue{ NumNodes: p.ClusterSize, RetransmitMult: retransmit, } messagesReceived := prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "alertmanager_cluster_messages_received_total", Help: "Total number of cluster messages received.", }, []string{"msg_type"}) messagesReceivedSize := prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "alertmanager_cluster_messages_received_size_total", Help: "Total size of cluster messages received.", }, []string{"msg_type"}) messagesSent := prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "alertmanager_cluster_messages_sent_total", Help: "Total number of cluster messages sent.", }, []string{"msg_type"}) messagesSentSize := prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "alertmanager_cluster_messages_sent_size_total", Help: "Total size of cluster messages sent.", }, []string{"msg_type"}) messagesPruned := prometheus.NewCounter(prometheus.CounterOpts{ Name: "alertmanager_cluster_messages_pruned_total", Help: "Total number of cluster messages pruned.", }) gossipClusterMembers := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "alertmanager_cluster_members", Help: "Number indicating current number of members in cluster.", }, func() float64 { return float64(p.ClusterSize()) }) peerPosition := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "alertmanager_peer_position", Help: "Position the Alertmanager instance believes it's in. The position determines a peer's behavior in the cluster.", }, func() float64 { return float64(p.Position()) }) healthScore := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "alertmanager_cluster_health_score", Help: "Health score of the cluster. Lower values are better and zero means 'totally healthy'.", }, func() float64 { return float64(p.mlist.GetHealthScore()) }) messagesQueued := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "alertmanager_cluster_messages_queued", Help: "Number of cluster messages which are queued.", }, func() float64 { return float64(bcast.NumQueued()) }) nodeAlive := prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "alertmanager_cluster_alive_messages_total", Help: "Total number of received alive messages.", }, []string{"peer"}, ) nodePingDuration := prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "alertmanager_cluster_pings_seconds", Help: "Histogram of latencies for ping messages.", Buckets: []float64{.005, .01, .025, .05, .1, .25, .5}, NativeHistogramBucketFactor: 1.1, NativeHistogramMaxBucketNumber: 100, NativeHistogramMinResetDuration: 1 * time.Hour, }, []string{"peer"}, ) messagesReceived.WithLabelValues(fullState) messagesReceivedSize.WithLabelValues(fullState) messagesReceived.WithLabelValues(update) messagesReceivedSize.WithLabelValues(update) messagesSent.WithLabelValues(fullState) messagesSentSize.WithLabelValues(fullState) messagesSent.WithLabelValues(update) messagesSentSize.WithLabelValues(update) reg.MustRegister(messagesReceived, messagesReceivedSize, messagesSent, messagesSentSize, gossipClusterMembers, peerPosition, healthScore, messagesQueued, messagesPruned, nodeAlive, nodePingDuration, ) d := &delegate{ logger: l, Peer: p, bcast: bcast, messagesReceived: messagesReceived, messagesReceivedSize: messagesReceivedSize, messagesSent: messagesSent, messagesSentSize: messagesSentSize, messagesPruned: messagesPruned, nodeAlive: nodeAlive, nodePingDuration: nodePingDuration, } go d.handleQueueDepth() return d } // NodeMeta retrieves meta-data about the current node when broadcasting an alive message. func (d delegate) NodeMeta(limit int) []byte { return []byte{} } // NotifyMsg is the callback invoked when a user-level gossip message is received. func (d delegate) NotifyMsg(b []byte) { d.messagesReceived.WithLabelValues(update).Inc() d.messagesReceivedSize.WithLabelValues(update).Add(float64(len(b))) var p clusterpb.Part if err := proto.Unmarshal(b, &p); err != nil { d.logger.Warn("decode broadcast", "err", err) return } d.mtx.RLock() s, ok := d.states[p.Key] d.mtx.RUnlock() if !ok { return } if err := s.Merge(p.Data); err != nil { d.logger.Warn("merge broadcast", "err", err, "key", p.Key) return } } // GetBroadcasts is called when user data messages can be broadcasted. func (d delegate) GetBroadcasts(overhead, limit int) [][]byte { msgs := d.bcast.GetBroadcasts(overhead, limit) d.messagesSent.WithLabelValues(update).Add(float64(len(msgs))) for _, m := range msgs { d.messagesSentSize.WithLabelValues(update).Add(float64(len(m))) } return msgs } // LocalState is called when gossip fetches local state. func (d delegate) LocalState(_ bool) []byte { d.mtx.RLock() defer d.mtx.RUnlock() all := &clusterpb.FullState{ Parts: make([]clusterpb.Part, 0, len(d.states)), } for key, s := range d.states { b, err := s.MarshalBinary() if err != nil { d.logger.Warn("encode local state", "err", err, "key", key) return nil } all.Parts = append(all.Parts, clusterpb.Part{Key: key, Data: b}) } b, err := proto.Marshal(all) if err != nil { d.logger.Warn("encode local state", "err", err) return nil } d.messagesSent.WithLabelValues(fullState).Inc() d.messagesSentSize.WithLabelValues(fullState).Add(float64(len(b))) return b } func (d delegate) MergeRemoteState(buf []byte, _ bool) { d.messagesReceived.WithLabelValues(fullState).Inc() d.messagesReceivedSize.WithLabelValues(fullState).Add(float64(len(buf))) var fs clusterpb.FullState if err := proto.Unmarshal(buf, &fs); err != nil { d.logger.Warn("merge remote state", "err", err) return } d.mtx.RLock() defer d.mtx.RUnlock() for _, p := range fs.Parts { s, ok := d.states[p.Key] if !ok { d.logger.Warn("unknown state key", "len", len(buf), "key", p.Key) continue } if err := s.Merge(p.Data); err != nil { d.logger.Warn("merge remote state", "err", err, "key", p.Key) return } } } // NotifyJoin is called if a peer joins the cluster. func (d delegate) NotifyJoin(n memberlist.Node) { d.logger.Debug("NotifyJoin", "node", n.Name, "addr", n.Address()) d.Peer.peerJoin(n) } // NotifyLeave is called if a peer leaves the cluster. func (d delegate) NotifyLeave(n memberlist.Node) { d.logger.Debug("NotifyLeave", "node", n.Name, "addr", n.Address()) d.Peer.peerLeave(n) } // NotifyUpdate is called if a cluster peer gets updated. func (d delegate) NotifyUpdate(n memberlist.Node) { d.logger.Debug("NotifyUpdate", "node", n.Name, "addr", n.Address()) d.Peer.peerUpdate(n) } // NotifyAlive implements the memberlist.AliveDelegate interface. func (d delegate) NotifyAlive(peer memberlist.Node) error { d.nodeAlive.WithLabelValues(peer.Name).Inc() return nil } // AckPayload implements the memberlist.PingDelegate interface. func (d delegate) AckPayload() []byte { return []byte{} } // NotifyPingComplete implements the memberlist.PingDelegate interface. func (d delegate) NotifyPingComplete(peer memberlist.Node, rtt time.Duration, payload []byte) { d.nodePingDuration.WithLabelValues(peer.Name).Observe(rtt.Seconds()) } // handleQueueDepth ensures that the queue doesn't grow unbounded by pruning // older messages at regular interval. func (d delegate) handleQueueDepth() { for { select { case <-d.stopc: return case <-time.After(15 time.Minute): n := d.bcast.NumQueued() if n > maxQueueSize { d.logger.Warn("dropping messages because too many are queued", "current", n, "limit", maxQueueSize) d.bcast.Prune(maxQueueSize) d.messagesPruned.Add(float64(n - maxQueueSize)) } } } }	9.22 KB	2025-05-25 12:27
📁 testdata	-	2025-05-25 12:27
📄 tls_config.go // Copyright 2020 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "fmt" "os" "path/filepath" "github.com/prometheus/common/config" "github.com/prometheus/exporter-toolkit/web" "gopkg.in/yaml.v2" ) type TLSTransportConfig struct { TLSServerConfig web.TLSConfig `yaml:"tls_server_config"` TLSClientConfig config.TLSConfig `yaml:"tls_client_config"` } func GetTLSTransportConfig(configPath string) (*TLSTransportConfig, error) { if configPath == "" { return nil, nil } bytes, err := os.ReadFile(configPath) if err != nil { return nil, err } cfg := &TLSTransportConfig{ TLSClientConfig: &config.TLSConfig{}, } if err := yaml.UnmarshalStrict(bytes, cfg); err != nil { return nil, err } if cfg.TLSServerConfig == nil { return nil, fmt.Errorf("missing 'tls_server_config' entry in the TLS configuration") } cfg.TLSServerConfig.SetDirectory(filepath.Dir(configPath)) cfg.TLSClientConfig.SetDirectory(filepath.Dir(configPath)) return cfg, nil }	1.5 KB	2025-05-25 12:27
📄 tls_connection.go // Copyright 2020 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "bufio" "crypto/tls" "encoding/binary" "errors" "fmt" "io" "net" "sync" "time" "github.com/gogo/protobuf/proto" "github.com/hashicorp/memberlist" "github.com/prometheus/alertmanager/cluster/clusterpb" ) const ( version = "v0.1.0" uint32length = 4 ) // tlsConn wraps net.Conn with connection pooling data. type tlsConn struct { mtx sync.Mutex connection net.Conn live bool } func dialTLSConn(addr string, timeout time.Duration, tlsConfig tls.Config) (tlsConn, error) { dialer := &net.Dialer{Timeout: timeout} conn, err := tls.DialWithDialer(dialer, network, addr, tlsConfig) if err != nil { return nil, err } return &tlsConn{ connection: conn, live: true, }, nil } func rcvTLSConn(conn net.Conn) tlsConn { return &tlsConn{ connection: conn, live: true, } } // Write writes a byte array into the connection. It returns the number of bytes written and an error. func (conn tlsConn) Write(b []byte) (int, error) { conn.mtx.Lock() defer conn.mtx.Unlock() n, err := conn.connection.Write(b) if err != nil { conn.live = false } return n, err } func (conn tlsConn) alive() bool { conn.mtx.Lock() defer conn.mtx.Unlock() return conn.live } func (conn tlsConn) getRawConn() net.Conn { conn.mtx.Lock() defer conn.mtx.Unlock() raw := conn.connection conn.live = false conn.connection = nil return raw } // writePacket writes all the bytes in one operation so no concurrent write happens in between. // It prefixes the message length. func (conn tlsConn) writePacket(fromAddr string, b []byte) error { msg, err := proto.Marshal( &clusterpb.MemberlistMessage{ Version: version, Kind: clusterpb.MemberlistMessage_PACKET, FromAddr: fromAddr, Msg: b, }, ) if err != nil { return fmt.Errorf("unable to marshal memeberlist packet message: %w", err) } buf := make([]byte, uint32length, uint32length+len(msg)) binary.LittleEndian.PutUint32(buf, uint32(len(msg))) _, err = conn.Write(append(buf, msg...)) return err } // writeStream simply signals that this is a stream connection by sending the connection type. func (conn tlsConn) writeStream() error { msg, err := proto.Marshal( &clusterpb.MemberlistMessage{ Version: version, Kind: clusterpb.MemberlistMessage_STREAM, }, ) if err != nil { return fmt.Errorf("unable to marshal memeberlist stream message: %w", err) } buf := make([]byte, uint32length, uint32length+len(msg)) binary.LittleEndian.PutUint32(buf, uint32(len(msg))) _, err = conn.Write(append(buf, msg...)) return err } // read returns a packet for packet connections or an error if there is one. // It returns nothing if the connection is meant to be streamed. func (conn tlsConn) read() (memberlist.Packet, error) { if conn.connection == nil { return nil, errors.New("nil connection") } conn.mtx.Lock() reader := bufio.NewReader(conn.connection) lenBuf := make([]byte, uint32length) _, err := io.ReadFull(reader, lenBuf) if err != nil { return nil, fmt.Errorf("error reading message length: %w", err) } msgLen := binary.LittleEndian.Uint32(lenBuf) msgBuf := make([]byte, msgLen) _, err = io.ReadFull(reader, msgBuf) conn.mtx.Unlock() if err != nil { return nil, fmt.Errorf("error reading message: %w", err) } pb := clusterpb.MemberlistMessage{} err = proto.Unmarshal(msgBuf, &pb) if err != nil { return nil, fmt.Errorf("error parsing message: %w", err) } if pb.Version != version { return nil, errors.New("tls memberlist message version incompatible") } switch pb.Kind { case clusterpb.MemberlistMessage_STREAM: return nil, nil case clusterpb.MemberlistMessage_PACKET: return toPacket(pb) default: return nil, errors.New("could not read from either stream or packet channel") } } func toPacket(pb clusterpb.MemberlistMessage) (memberlist.Packet, error) { addr, err := net.ResolveTCPAddr(network, pb.FromAddr) if err != nil { return nil, fmt.Errorf("error parsing packet sender address: %w", err) } return &memberlist.Packet{ Buf: pb.Msg, From: addr, Timestamp: time.Now(), }, nil } func (conn tlsConn) Close() error { conn.mtx.Lock() defer conn.mtx.Unlock() conn.live = false if conn.connection == nil { return nil } return conn.connection.Close() }	4.79 KB	2025-05-25 12:27
📄 tls_connection_test.go // Copyright 2020 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "errors" "net" "testing" "time" "github.com/stretchr/testify/require" ) func TestWriteStream(t testing.T) { w, r := net.Pipe() conn := &tlsConn{ connection: w, } defer r.Close() go func() { conn.writeStream() w.Close() }() packet, err := rcvTLSConn(r).read() require.NoError(t, err) require.Nil(t, packet) } func TestWritePacket(t testing.T) { testCases := []struct { fromAddr string msg string }{ {fromAddr: "127.0.0.1:8001", msg: ""}, {fromAddr: "10.0.0.4:9094", msg: "hello"}, {fromAddr: "127.0.0.1:8001", msg: "0"}, } for _, tc := range testCases { w, r := net.Pipe() defer r.Close() go func() { conn := &tlsConn{connection: w} conn.writePacket(tc.fromAddr, []byte(tc.msg)) w.Close() }() packet, err := rcvTLSConn(r).read() require.NoError(t, err) require.Equal(t, tc.msg, string(packet.Buf)) require.Equal(t, tc.fromAddr, packet.From.String()) } } func TestRead_Nil(t testing.T) { packet, err := (&tlsConn{}).read() require.Nil(t, packet) require.Error(t, err) } func TestTLSConn_Close(t testing.T) { testCases := []string{ "foo", "bar", } for _, tc := range testCases { c := &tlsConn{ connection: &mockConn{ errMsg: tc, }, live: true, } err := c.Close() require.Equal(t, errors.New(tc), err, tc) require.False(t, c.alive()) require.True(t, c.connection.(mockConn).closed) } } type mockConn struct { closed bool errMsg string } func (m mockConn) Read(b []byte) (n int, err error) { panic("implement me") } func (m mockConn) Write(b []byte) (n int, err error) { panic("implement me") } func (m mockConn) Close() error { m.closed = true return errors.New(m.errMsg) } func (m mockConn) LocalAddr() net.Addr { panic("implement me") } func (m mockConn) RemoteAddr() net.Addr { panic("implement me") } func (m mockConn) SetDeadline(t time.Time) error { panic("implement me") } func (m mockConn) SetReadDeadline(t time.Time) error { panic("implement me") } func (m *mockConn) SetWriteDeadline(t time.Time) error { panic("implement me") }	2.63 KB	2025-05-25 12:27
📄 tls_transport.go // Copyright 2020 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // Forked from https://github.com/mxinden/memberlist-tls-transport. // Implements Transport interface so that all gossip communications occur via TLS over TCP. package cluster import ( "context" "crypto/tls" "errors" "fmt" "log/slog" "net" "strings" "time" "github.com/hashicorp/go-sockaddr" "github.com/hashicorp/memberlist" "github.com/prometheus/client_golang/prometheus" common "github.com/prometheus/common/config" "github.com/prometheus/exporter-toolkit/web" ) const ( metricNamespace = "alertmanager" metricSubsystem = "tls_transport" network = "tcp" ) // TLSTransport is a Transport implementation that uses TLS over TCP for both // packet and stream operations. type TLSTransport struct { ctx context.Context cancel context.CancelFunc logger slog.Logger bindAddr string bindPort int done chan struct{} listener net.Listener packetCh chan memberlist.Packet streamCh chan net.Conn connPool connectionPool tlsServerCfg tls.Config tlsClientCfg tls.Config packetsSent prometheus.Counter packetsRcvd prometheus.Counter streamsSent prometheus.Counter streamsRcvd prometheus.Counter readErrs prometheus.Counter writeErrs prometheus.CounterVec } // NewTLSTransport returns a TLS transport with the given configuration. // On successful initialization, a tls listener will be created and listening. // A valid bindAddr is required. If bindPort == 0, the system will assign // a free port automatically. func NewTLSTransport( ctx context.Context, logger slog.Logger, reg prometheus.Registerer, bindAddr string, bindPort int, cfg TLSTransportConfig, ) (TLSTransport, error) { if cfg == nil { return nil, errors.New("must specify TLSTransportConfig") } tlsServerCfg, err := web.ConfigToTLSConfig(cfg.TLSServerConfig) if err != nil { return nil, fmt.Errorf("invalid TLS server config: %w", err) } tlsClientCfg, err := common.NewTLSConfig(cfg.TLSClientConfig) if err != nil { return nil, fmt.Errorf("invalid TLS client config: %w", err) } ip := net.ParseIP(bindAddr) if ip == nil { return nil, fmt.Errorf("invalid bind address \"%s\"", bindAddr) } addr := &net.TCPAddr{IP: ip, Port: bindPort} listener, err := tls.Listen(network, addr.String(), tlsServerCfg) if err != nil { return nil, fmt.Errorf("failed to start TLS listener on %q port %d: %w", bindAddr, bindPort, err) } connPool, err := newConnectionPool(tlsClientCfg) if err != nil { return nil, fmt.Errorf("failed to initialize tls transport connection pool: %w", err) } ctx, cancel := context.WithCancel(ctx) t := &TLSTransport{ ctx: ctx, cancel: cancel, logger: logger, bindAddr: bindAddr, bindPort: bindPort, done: make(chan struct{}), listener: listener, packetCh: make(chan memberlist.Packet), streamCh: make(chan net.Conn), connPool: connPool, tlsServerCfg: tlsServerCfg, tlsClientCfg: tlsClientCfg, } t.registerMetrics(reg) go func() { t.listen() close(t.done) }() return t, nil } // FinalAdvertiseAddr is given the user's configured values (which // might be empty) and returns the desired IP and port to advertise to // the rest of the cluster. func (t TLSTransport) FinalAdvertiseAddr(ip string, port int) (net.IP, int, error) { var advertiseAddr net.IP var advertisePort int if ip != "" { advertiseAddr = net.ParseIP(ip) if advertiseAddr == nil { return nil, 0, fmt.Errorf("failed to parse advertise address %q", ip) } if ip4 := advertiseAddr.To4(); ip4 != nil { advertiseAddr = ip4 } advertisePort = port } else { if t.bindAddr == "0.0.0.0" { // Otherwise, if we're not bound to a specific IP, let's // use a suitable private IP address. var err error ip, err = sockaddr.GetPrivateIP() if err != nil { return nil, 0, fmt.Errorf("failed to get interface addresses: %w", err) } if ip == "" { return nil, 0, fmt.Errorf("no private IP address found, and explicit IP not provided") } advertiseAddr = net.ParseIP(ip) if advertiseAddr == nil { return nil, 0, fmt.Errorf("failed to parse advertise address: %q", ip) } } else { advertiseAddr = t.listener.Addr().(net.TCPAddr).IP } advertisePort = t.GetAutoBindPort() } return advertiseAddr, advertisePort, nil } // PacketCh returns a channel that can be read to receive incoming // packets from other peers. func (t TLSTransport) PacketCh() <-chan memberlist.Packet { return t.packetCh } // StreamCh returns a channel that can be read to handle incoming stream // connections from other peers. func (t TLSTransport) StreamCh() <-chan net.Conn { return t.streamCh } // Shutdown is called when memberlist is shutting down; this gives the // TLS Transport a chance to clean up the listener and other goroutines. func (t TLSTransport) Shutdown() error { t.logger.Debug("shutting down tls transport") t.cancel() err := t.listener.Close() t.connPool.shutdown() <-t.done return err } // WriteTo is a packet-oriented interface that borrows a connection // from the pool, and writes to it. It also returns a timestamp of when // the packet was written. func (t TLSTransport) WriteTo(b []byte, addr string) (time.Time, error) { conn, err := t.connPool.borrowConnection(addr, DefaultTCPTimeout) if err != nil { t.writeErrs.WithLabelValues("packet").Inc() return time.Now(), fmt.Errorf("failed to dial: %w", err) } fromAddr := t.listener.Addr().String() err = conn.writePacket(fromAddr, b) if err != nil { t.writeErrs.WithLabelValues("packet").Inc() return time.Now(), fmt.Errorf("failed to write packet: %w", err) } t.packetsSent.Add(float64(len(b))) return time.Now(), nil } // DialTimeout is used to create a connection that allows memberlist // to perform two-way communications with a peer. func (t TLSTransport) DialTimeout(addr string, timeout time.Duration) (net.Conn, error) { conn, err := dialTLSConn(addr, timeout, t.tlsClientCfg) if err != nil { t.writeErrs.WithLabelValues("stream").Inc() return nil, fmt.Errorf("failed to dial: %w", err) } err = conn.writeStream() netConn := conn.getRawConn() if err != nil { t.writeErrs.WithLabelValues("stream").Inc() return netConn, fmt.Errorf("failed to create stream connection: %w", err) } t.streamsSent.Inc() return netConn, nil } // GetAutoBindPort returns the bind port that was automatically given by the system // if a bindPort of 0 was specified during instantiation. func (t TLSTransport) GetAutoBindPort() int { return t.listener.Addr().(net.TCPAddr).Port } // listen starts up multiple handlers accepting concurrent connections. func (t TLSTransport) listen() { for { select { case <-t.ctx.Done(): return default: conn, err := t.listener.Accept() if err != nil { // The error "use of closed network connection" is returned when the listener is closed. // It is not exported in a more reasonable way. See https://github.com/golang/go/issues/4373. if strings.Contains(err.Error(), "use of closed network connection") { return } t.readErrs.Inc() t.logger.Debug("error accepting connection", "err", err) } else { go t.handle(conn) } } } } func (t TLSTransport) handle(conn net.Conn) { for { packet, err := rcvTLSConn(conn).read() if err != nil { t.logger.Debug("error reading from connection", "err", err) t.readErrs.Inc() return } select { case <-t.ctx.Done(): return default: if packet != nil { n := len(packet.Buf) t.packetCh <- packet t.packetsRcvd.Add(float64(n)) } else { t.streamCh <- conn t.streamsRcvd.Inc() return } } } } func (t *TLSTransport) registerMetrics(reg prometheus.Registerer) { t.packetsSent = prometheus.NewCounter( prometheus.CounterOpts{ Namespace: metricNamespace, Subsystem: metricSubsystem, Name: "packet_bytes_sent_total", Help: "The number of packet bytes sent to outgoing connections (excluding internal metadata).", }, ) t.packetsRcvd = prometheus.NewCounter( prometheus.CounterOpts{ Namespace: metricNamespace, Subsystem: metricSubsystem, Name: "packet_bytes_received_total", Help: "The number of packet bytes received from incoming connections (excluding internal metadata).", }, ) t.streamsSent = prometheus.NewCounter( prometheus.CounterOpts{ Namespace: metricNamespace, Subsystem: metricSubsystem, Name: "stream_connections_sent_total", Help: "The number of stream connections sent.", }, ) t.streamsRcvd = prometheus.NewCounter( prometheus.CounterOpts{ Namespace: metricNamespace, Subsystem: metricSubsystem, Name: "stream_connections_received_total", Help: "The number of stream connections received.", }, ) t.readErrs = prometheus.NewCounter( prometheus.CounterOpts{ Namespace: metricNamespace, Subsystem: metricSubsystem, Name: "read_errors_total", Help: "The number of errors encountered while reading from incoming connections.", }, ) t.writeErrs = prometheus.NewCounterVec( prometheus.CounterOpts{ Namespace: metricNamespace, Subsystem: metricSubsystem, Name: "write_errors_total", Help: "The number of errors encountered while writing to outgoing connections.", }, []string{"connection_type"}, ) if reg != nil { reg.MustRegister(t.packetsSent) reg.MustRegister(t.packetsRcvd) reg.MustRegister(t.streamsSent) reg.MustRegister(t.streamsRcvd) reg.MustRegister(t.readErrs) reg.MustRegister(t.writeErrs) } }	9.96 KB	2025-05-25 12:27
📄 tls_transport_test.go // Copyright 2020 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package cluster import ( "bufio" "bytes" context2 "context" "fmt" "io" "net" "sync" "testing" "time" "github.com/prometheus/common/promslog" "github.com/stretchr/testify/require" ) var logger = promslog.NewNopLogger() func freeport() int { lis, _ := net.Listen(network, "127.0.0.1:0") defer lis.Close() return lis.Addr().(net.TCPAddr).Port } func newTLSTransport(file, address string, port int) (TLSTransport, error) { cfg, err := GetTLSTransportConfig(file) if err != nil { return nil, err } return NewTLSTransport(context2.Background(), promslog.NewNopLogger(), nil, address, port, cfg) } func TestNewTLSTransport(t testing.T) { port := freeport() for _, tc := range []struct { bindAddr string bindPort int tlsConfFile string err string }{ { err: "must specify TLSTransportConfig", }, { tlsConfFile: "testdata/empty_tls_config.yml", err: "missing 'tls_server_config' entry in the TLS configuration", }, { tlsConfFile: "testdata/tls_config_with_missing_server.yml", err: "missing 'tls_server_config' entry in the TLS configuration", }, { err: "invalid bind address \"\"", tlsConfFile: "testdata/tls_config_node1.yml", }, { bindAddr: "abc123", err: "invalid bind address \"abc123\"", tlsConfFile: "testdata/tls_config_node1.yml", }, { bindAddr: localhost, bindPort: 0, tlsConfFile: "testdata/tls_config_node1.yml", }, { bindAddr: localhost, bindPort: port, tlsConfFile: "testdata/tls_config_node2.yml", }, { tlsConfFile: "testdata/tls_config_with_missing_client.yml", bindAddr: localhost, }, } { t.Run("", func(t testing.T) { transport, err := newTLSTransport(tc.tlsConfFile, tc.bindAddr, tc.bindPort) if len(tc.err) > 0 { require.Error(t, err) require.Equal(t, tc.err, err.Error()) return } defer transport.Shutdown() require.NoError(t, err) require.Equal(t, tc.bindAddr, transport.bindAddr) require.Equal(t, tc.bindPort, transport.bindPort) require.NotNil(t, transport.listener) }) } } const localhost = "127.0.0.1" func TestFinalAdvertiseAddr(t testing.T) { ports := [...]int{freeport(), freeport(), freeport()} testCases := []struct { bindAddr string bindPort int inputIP string inputPort int expectedIP string expectedPort int expectedError string }{ {bindAddr: localhost, bindPort: ports[0], inputIP: "10.0.0.5", inputPort: 54231, expectedIP: "10.0.0.5", expectedPort: 54231}, {bindAddr: localhost, bindPort: ports[1], inputIP: "invalid", inputPort: 54231, expectedError: "failed to parse advertise address \"invalid\""}, {bindAddr: "0.0.0.0", bindPort: 0, inputIP: "", inputPort: 0, expectedIP: "random"}, {bindAddr: localhost, bindPort: 0, inputIP: "", inputPort: 0, expectedIP: localhost}, {bindAddr: localhost, bindPort: ports[2], inputIP: "", inputPort: 0, expectedIP: localhost, expectedPort: ports[2]}, } for _, tc := range testCases { tlsConf := loadTLSTransportConfig(t, "testdata/tls_config_node1.yml") transport, err := NewTLSTransport(context2.Background(), logger, nil, tc.bindAddr, tc.bindPort, tlsConf) require.NoError(t, err) ip, port, err := transport.FinalAdvertiseAddr(tc.inputIP, tc.inputPort) if len(tc.expectedError) > 0 { require.Equal(t, tc.expectedError, err.Error()) } else { require.NoError(t, err) if tc.expectedPort == 0 { require.Less(t, tc.expectedPort, port) } else { require.Equal(t, tc.expectedPort, port) } if tc.expectedIP == "random" { require.NotNil(t, ip) } else { require.Equal(t, tc.expectedIP, ip.String()) } } transport.Shutdown() } } func TestWriteTo(t testing.T) { tlsConf1 := loadTLSTransportConfig(t, "testdata/tls_config_node1.yml") t1, _ := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf1) defer t1.Shutdown() tlsConf2 := loadTLSTransportConfig(t, "testdata/tls_config_node2.yml") t2, _ := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf2) defer t2.Shutdown() from := fmt.Sprintf("%s:%d", t1.bindAddr, t1.GetAutoBindPort()) to := fmt.Sprintf("%s:%d", t2.bindAddr, t2.GetAutoBindPort()) sent := []byte(("test packet")) _, err := t1.WriteTo(sent, to) require.NoError(t, err) packet := <-t2.PacketCh() require.Equal(t, sent, packet.Buf) require.Equal(t, from, packet.From.String()) } func BenchmarkWriteTo(b testing.B) { tlsConf1 := loadTLSTransportConfig(b, "testdata/tls_config_node1.yml") t1, _ := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf1) defer t1.Shutdown() tlsConf2 := loadTLSTransportConfig(b, "testdata/tls_config_node2.yml") t2, _ := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf2) defer t2.Shutdown() b.ResetTimer() from := fmt.Sprintf("%s:%d", t1.bindAddr, t1.GetAutoBindPort()) to := fmt.Sprintf("%s:%d", t2.bindAddr, t2.GetAutoBindPort()) sent := []byte(("test packet")) _, err := t1.WriteTo(sent, to) require.NoError(b, err) packet := <-t2.PacketCh() require.Equal(b, sent, packet.Buf) require.Equal(b, from, packet.From.String()) } func TestDialTimeout(t testing.T) { tlsConf1 := loadTLSTransportConfig(t, "testdata/tls_config_node1.yml") t1, err := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf1) require.NoError(t, err) defer t1.Shutdown() tlsConf2 := loadTLSTransportConfig(t, "testdata/tls_config_node2.yml") t2, err := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf2) require.NoError(t, err) defer t2.Shutdown() addr := fmt.Sprintf("%s:%d", t2.bindAddr, t2.GetAutoBindPort()) from, err := t1.DialTimeout(addr, 5time.Second) require.NoError(t, err) defer from.Close() var to net.Conn var wg sync.WaitGroup wg.Add(1) go func() { to = <-t2.StreamCh() wg.Done() }() sent := []byte(("test stream")) m, err := from.Write(sent) require.NoError(t, err) require.Positive(t, m) wg.Wait() reader := bufio.NewReader(to) buf := make([]byte, len(sent)) n, err := io.ReadFull(reader, buf) require.NoError(t, err) require.Len(t, sent, n) require.Equal(t, sent, buf) } func TestShutdown(t testing.T) { var buf bytes.Buffer promslogConfig := &promslog.Config{Writer: &buf} logger := promslog.New(promslogConfig) // Set logger to debug, otherwise it won't catch some logging from `Shutdown()` method. _ = promslogConfig.Level.Set("debug") tlsConf1 := loadTLSTransportConfig(t, "testdata/tls_config_node1.yml") t1, _ := NewTLSTransport(context2.Background(), logger, nil, "127.0.0.1", 0, tlsConf1) // Sleeping to make sure listeners have started and can subsequently be shut down gracefully. time.Sleep(500 * time.Millisecond) err := t1.Shutdown() require.NoError(t, err) require.NotContains(t, buf.String(), "use of closed network connection") require.Contains(t, buf.String(), "shutting down tls transport") } func loadTLSTransportConfig(tb testing.TB, filename string) *TLSTransportConfig { tb.Helper() config, err := GetTLSTransportConfig(filename) if err != nil { tb.Fatal(err) } return config }	7.6 KB	2025-05-25 12:27

Terminal