diff --git a/cmd/internal/storage/redis/redis.go b/cmd/internal/storage/redis/redis.go index 1ccec87eb6..1d9ac3fd26 100644 --- a/cmd/internal/storage/redis/redis.go +++ b/cmd/internal/storage/redis/redis.go @@ -16,6 +16,7 @@ package redis import ( "encoding/json" + "flag" "os" "sync" "time" @@ -56,10 +57,23 @@ func new() (storage.StorageDriver, error) { hostname, *storage.ArgDbName, *storage.ArgDbHost, + configuredRedisPassword(), *storage.ArgDbBufferDuration, ) } +func configuredRedisPassword() string { + password := "" + flag.Visit(func(f *flag.Flag) { + if f.Name == "storage_driver_password" { + // The shared storage password flag defaults to "root" for other + // backends. Redis should authenticate only when the flag is set. + password = *storage.ArgDbPassword + } + }) + return password +} + func (s *redisStorage) defaultReadyToFlush() bool { return time.Since(s.lastWrite) >= s.bufferDuration } @@ -105,7 +119,8 @@ func (s *redisStorage) AddStats(cInfo *info.ContainerInfo, stats *info.Container return nil } // We use redis's "LPUSH" to push the data to the redis - return s.conn.Send("LPUSH", s.redisKey, seriesToFlush) + _, err := s.conn.Do("LPUSH", s.redisKey, seriesToFlush) + return err } func (s *redisStorage) Close() error { @@ -120,10 +135,15 @@ func (s *redisStorage) Close() error { func newStorage( machineName, redisKey, - redisHost string, + redisHost, + redisPassword string, bufferDuration time.Duration, ) (storage.StorageDriver, error) { - conn, err := redis.Dial("tcp", redisHost) + var dialOptions []redis.DialOption + if redisPassword != "" { + dialOptions = append(dialOptions, redis.DialPassword(redisPassword)) + } + conn, err := redis.Dial("tcp", redisHost, dialOptions...) if err != nil { return nil, err } diff --git a/cmd/internal/storage/redis/redis_test.go b/cmd/internal/storage/redis/redis_test.go new file mode 100644 index 0000000000..1d18dc74b8 --- /dev/null +++ b/cmd/internal/storage/redis/redis_test.go @@ -0,0 +1,201 @@ +// Copyright 2026 Google Inc. All Rights Reserved. +// +// 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 redis + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "net" + "strconv" + "strings" + "testing" + "time" + + info "github.com/google/cadvisor/info/v1" +) + +func TestAddStatsAuthenticatesWhenPasswordConfigured(t *testing.T) { + commands, address, stop := startFakeRedis(t) + defer stop() + + driver, err := newStorage( + "test-machine", + "cadvisor", + address, + "redisPswrd", + 0, + ) + if err != nil { + t.Fatalf("newStorage() error = %v", err) + } + defer driver.Close() + + storage := driver.(*redisStorage) + storage.readyToFlush = func() bool { return true } + + if err := storage.AddStats( + &info.ContainerInfo{ + ContainerReference: info.ContainerReference{ + Name: "/docker/test-container", + }, + }, + &info.ContainerStats{Timestamp: time.Unix(1, 0)}, + ); err != nil { + t.Fatalf("AddStats() error = %v", err) + } + + assertCommand(t, <-commands, "AUTH", "redisPswrd") + lpush := <-commands + if len(lpush) != 3 { + t.Fatalf("LPUSH command length = %d, want 3: %#v", len(lpush), lpush) + } + assertCommand(t, lpush[:2], "LPUSH", "cadvisor") + + var detail detailSpec + if err := json.Unmarshal([]byte(lpush[2]), &detail); err != nil { + t.Fatalf("LPUSH payload is not detailSpec JSON: %v", err) + } + if detail.MachineName != "test-machine" { + t.Fatalf("MachineName = %q, want test-machine", detail.MachineName) + } + if detail.ContainerName != "/docker/test-container" { + t.Fatalf( + "ContainerName = %q, want /docker/test-container", + detail.ContainerName, + ) + } +} + +func TestAddStatsDoesNotAuthenticateWithoutPassword(t *testing.T) { + commands, address, stop := startFakeRedis(t) + defer stop() + + driver, err := newStorage("test-machine", "cadvisor", address, "", 0) + if err != nil { + t.Fatalf("newStorage() error = %v", err) + } + defer driver.Close() + + storage := driver.(*redisStorage) + storage.readyToFlush = func() bool { return true } + + if err := storage.AddStats( + &info.ContainerInfo{ + ContainerReference: info.ContainerReference{ + Name: "/docker/test-container", + }, + }, + &info.ContainerStats{Timestamp: time.Unix(1, 0)}, + ); err != nil { + t.Fatalf("AddStats() error = %v", err) + } + + lpush := <-commands + assertCommand(t, lpush[:2], "LPUSH", "cadvisor") +} + +func startFakeRedis(t *testing.T) (<-chan []string, string, func()) { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("net.Listen() error = %v", err) + } + + commands := make(chan []string, 4) + done := make(chan struct{}) + go func() { + defer close(done) + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + reader := bufio.NewReader(conn) + for { + command, err := readRESPCommand(reader) + if err != nil { + return + } + commands <- command + switch strings.ToUpper(command[0]) { + case "AUTH": + fmt.Fprint(conn, "+OK\r\n") + case "LPUSH": + fmt.Fprint(conn, ":1\r\n") + default: + fmt.Fprintf(conn, "-ERR unsupported command %s\r\n", command[0]) + } + } + }() + + stop := func() { + listener.Close() + <-done + } + return commands, listener.Addr().String(), stop +} + +func readRESPCommand(reader *bufio.Reader) ([]string, error) { + line, err := reader.ReadString('\n') + if err != nil { + return nil, err + } + line = strings.TrimSuffix(line, "\r\n") + if !strings.HasPrefix(line, "*") { + return nil, fmt.Errorf("expected array, got %q", line) + } + count, err := strconv.Atoi(strings.TrimPrefix(line, "*")) + if err != nil { + return nil, err + } + + command := make([]string, 0, count) + for i := 0; i < count; i++ { + lengthLine, err := reader.ReadString('\n') + if err != nil { + return nil, err + } + lengthLine = strings.TrimSuffix(lengthLine, "\r\n") + if !strings.HasPrefix(lengthLine, "$") { + return nil, fmt.Errorf("expected bulk string, got %q", lengthLine) + } + length, err := strconv.Atoi(strings.TrimPrefix(lengthLine, "$")) + if err != nil { + return nil, err + } + buf := make([]byte, length+2) + if _, err := io.ReadFull(reader, buf); err != nil { + return nil, err + } + command = append(command, string(buf[:length])) + } + return command, nil +} + +func assertCommand(t *testing.T, command []string, want ...string) { + t.Helper() + if len(command) != len(want) { + t.Fatalf("command length = %d, want %d: %#v", len(command), len(want), command) + } + for i := range want { + if command[i] != want[i] { + t.Fatalf("command[%d] = %q, want %q: %#v", i, command[i], want[i], command) + } + } +}