frontend: Use net.JoinHostPort to support IPv6 addresses

Fixes an issue when using IPv6 where IPv6 addresses are not properly
joined with ports resulting in the ring attempting to incorrectly
use `::1:7946` instead of the proper `[::1]:7946` notation.

With this change it's fully possible to use Loki within an IPv6 only
network.

Signed-off-by: Matthew Penner <[email protected]>

CHANGELOG.md

@@ -51,6 +51,7 @@
 * [10814](https://github.com/grafana/loki/pull/10814) **shantanualsi,kaviraj** Upgrade prometheus to v0.47.1 and dskit
 * [10959](https://github.com/grafana/loki/pull/10959) **slim-bean** introduce a backoff wait on subquery retries.
 * [11121](https://github.com/grafana/loki/pull/11121) **periklis** Ensure all lifecycler cfgs ref a valid IPv6 addr and port combination
+* [10650](https://github.com/grafana/loki/pull/10650) **matthewpi** Ensure the frontend uses a valid IPv6 addr and port combination
 
 #### Promtail
 

clients/pkg/promtail/discovery/consulagent/consul.go

@@ -8,7 +8,6 @@ package consulagent
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"net"
 	"net/http"
 	"strconv"
@@ -527,9 +526,9 @@ func (srv *consulService) watch(ctx context.Context, ch chan<- []*targetgroup.Gr
 		// since the service may be registered remotely through a different node.
 		var addr string
 		if srvCheck.Service.Address != "" {
-			addr = net.JoinHostPort(srvCheck.Service.Address, fmt.Sprintf("%d", srvCheck.Service.Port))
+			addr = net.JoinHostPort(srvCheck.Service.Address, strconv.Itoa(srvCheck.Service.Port))
 		} else {
-			addr = net.JoinHostPort(member.Addr, fmt.Sprintf("%d", srvCheck.Service.Port))
+			addr = net.JoinHostPort(member.Addr, strconv.Itoa(srvCheck.Service.Port))
 		}
 
 		labels := model.LabelSet{
@@ -560,7 +559,7 @@ func (srv *consulService) watch(ctx context.Context, ch chan<- []*targetgroup.Gr
 		// Add all key/value pairs from the service's tagged addresses as their own labels.
 		for k, v := range srvCheck.Service.TaggedAddresses {
 			name := strutil.SanitizeLabelName(k)
-			address := fmt.Sprintf("%s:%d", v.Address, v.Port)
+			address := net.JoinHostPort(v.Address, strconv.Itoa(v.Port))
 			labels[taggedAddressesLabel+model.LabelName(name)] = model.LabelValue(address)
 		}
 

pkg/loki/modules.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"hash/fnv"
 	"math"
+	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -866,7 +867,7 @@ func (t *Loki) compactorAddress() (string, bool, error) {
 	legacyReadMode := t.Cfg.LegacyReadTarget && t.Cfg.isModuleEnabled(Read)
 	if t.Cfg.isModuleEnabled(All) || legacyReadMode || t.Cfg.isModuleEnabled(Backend) {
 		// In single binary or read modes, this module depends on Server
-		return fmt.Sprintf("%s:%d", t.Cfg.Server.GRPCListenAddress, t.Cfg.Server.GRPCListenPort), true, nil
+		return net.JoinHostPort(t.Cfg.Server.GRPCListenAddress, strconv.Itoa(t.Cfg.Server.GRPCListenPort)), true, nil
 	}
 
 	if t.Cfg.Common.CompactorAddress == "" && t.Cfg.Common.CompactorGRPCAddress == "" {

pkg/lokifrontend/frontend/v2/frontend.go

@@ -5,7 +5,9 @@ import (
 	"flag"
 	"fmt"
 	"math/rand"
+	"net"
 	"net/http"
+	"strconv"
 	"sync"
 	"time"
 
@@ -136,7 +138,7 @@ type enqueueResult struct {
 func NewFrontend(cfg Config, ring ring.ReadRing, log log.Logger, reg prometheus.Registerer, codec transport.Codec, metricsNamespace string) (*Frontend, error) {
 	requestsCh := make(chan *frontendRequest)
 
-	schedulerWorkers, err := newFrontendSchedulerWorkers(cfg, fmt.Sprintf("%s:%d", cfg.Addr, cfg.Port), ring, requestsCh, log)
+	schedulerWorkers, err := newFrontendSchedulerWorkers(cfg, net.JoinHostPort(cfg.Addr, strconv.Itoa(cfg.Port)), ring, requestsCh, log)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/worker_service.go

@@ -1,7 +1,8 @@
 package querier
 
 import (
-	"fmt"
+	"net"
+	"strconv"
 
 	"github.com/go-kit/log/level"
 	"github.com/grafana/dskit/ring"
@@ -89,7 +90,7 @@ func InitWorkerService(
 		if cfg.GrpcListenAddress != "" {
 			listenAddress = cfg.GrpcListenAddress
 		}
-		address := fmt.Sprintf("%s:%d", listenAddress, cfg.GrpcListenPort)
+		address := net.JoinHostPort(listenAddress, strconv.Itoa(cfg.GrpcListenPort))
 		level.Warn(util_log.Logger).Log(
 			"msg", "Worker address is empty, attempting automatic worker configuration. If queries are unresponsive consider configuring the worker explicitly.",
 			"address", address)

pkg/storage/chunk/cache/memcached_client.go

@@ -3,9 +3,9 @@ package cache
 import (
 	"context"
 	"flag"
-	"fmt"
 	"net"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -254,7 +254,7 @@ func (c *memcachedClient) updateMemcacheServers() error {
 			return err
 		}
 		for _, srv := range addrs {
-			servers = append(servers, fmt.Sprintf("%s:%d", srv.Target, srv.Port))
+			servers = append(servers, net.JoinHostPort(srv.Target, strconv.Itoa(int(srv.Port))))
 		}
 	}