feature: add handling for Reserved IPv6 assignment and unassignment

Makefile

@@ -238,3 +238,4 @@ mockgen:
 	GOOS=linux mockgen -source=internal/netutil/tcp_sniffer_helper_linux.go -package=mocks -destination=internal/netutil/internal/mocks/dependent_functions_mock.go
 	mockgen -source=internal/metadata/updater/updater.go -package=updater -destination=internal/metadata/updater/updater_mocks.go
 	mockgen -destination=internal/metadata/updater/readcloser_mocks.go -package=updater -build_flags=--mod=mod io ReadCloser
+	mockgen -typed -source=internal/reservedipv6/reserved_ipv6.go -package=reservedipv6 -destination=internal/reservedipv6/mocks.go

cmd/agent/main.go

@@ -12,12 +12,15 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/jsimonetti/rtnetlink/v2"
+
 	"github.com/digitalocean/droplet-agent/internal/config"
 	"github.com/digitalocean/droplet-agent/internal/log"
 	"github.com/digitalocean/droplet-agent/internal/metadata"
 	"github.com/digitalocean/droplet-agent/internal/metadata/actioner"
 	"github.com/digitalocean/droplet-agent/internal/metadata/updater"
 	"github.com/digitalocean/droplet-agent/internal/metadata/watcher"
+	"github.com/digitalocean/droplet-agent/internal/reservedipv6"
 	"github.com/digitalocean/droplet-agent/internal/sysaccess"
 )
 
@@ -51,19 +54,36 @@ func main() {
 		log.Fatal("failed to initialize SSHManager: %v", err)
 	}
 
-	doManagedKeysActioner := actioner.NewDOManagedKeysActioner(sshMgr)
+	// create the watcher
 	metadataWatcher := newMetadataWatcher(&watcher.Conf{SSHPort: sshMgr.SSHDPort()})
-	metadataWatcher.RegisterActioner(doManagedKeysActioner)
-	infoUpdater := updater.NewAgentInfoUpdater()
 
-	// monitor sshd_config
+	// ssh managed keys
+	doManagedKeysActioner := actioner.NewDOManagedKeysActioner(sshMgr)
+	metadataWatcher.RegisterActioner(doManagedKeysActioner)
 	go mustMonitorSSHDConfig(sshMgr)
-
-	// Launch background jobs
 	bgJobsCtx, bgJobsCancel := context.WithCancel(context.Background())
 	go bgJobsRemoveExpiredDOTTYKeys(bgJobsCtx, sshMgr, cfg.AuthorizedKeysCheckInterval)
 
+	// reserved ipv6
+	if cfg.ManageReservedIPv6 {
+		log.Info("Reserved IPv6 management enabled")
+		conn, err := rtnetlink.Dial(nil)
+		if err != nil {
+			log.Fatal("failed to create netlink client: %v", err)
+		}
+		defer conn.Close()
+
+		rip6Manager, err := reservedipv6.NewManager(conn)
+		if err != nil {
+			log.Fatal("Failed to create Reserved IPv6 manager: %v", err)
+		}
+
+		rip6Actioner := actioner.NewReservedIPv6Actioner(rip6Manager)
+		metadataWatcher.RegisterActioner(rip6Actioner)
+	}
+
 	// handle shutdown
+	infoUpdater := updater.NewAgentInfoUpdater()
 	go handleShutdown(bgJobsCancel, metadataWatcher, infoUpdater, sshMgr)
 
 	// report agent status and ssh info

go.mod

@@ -4,6 +4,8 @@ go 1.22
 
 require (
 	github.com/fsnotify/fsnotify v1.7.0
+	github.com/jsimonetti/rtnetlink/v2 v2.0.2
+	github.com/mdlayher/netlink v1.7.2
 	github.com/opencontainers/selinux v1.11.0
 	github.com/peterbourgon/ff/v3 v3.4.0
 	go.uber.org/mock v0.4.0
@@ -13,3 +15,9 @@ require (
 	golang.org/x/sys v0.26.0
 	golang.org/x/time v0.7.0
 )
+
+require (
+	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/josharian/native v1.1.0 // indirect
+	github.com/mdlayher/socket v0.4.1 // indirect
+)

go.sum

@@ -1,5 +1,17 @@
+github.com/cilium/ebpf v0.12.3 h1:8ht6F9MquybnY97at+VDZb3eQQr8ev79RueWeVaEcG4=
+github.com/cilium/ebpf v0.12.3/go.mod h1:TctK1ivibvI3znr66ljgi4hqOT8EYQjz1KWBfb1UVgM=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
 github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/josharian/native v1.1.0 h1:uuaP0hAbW7Y4l0ZRQ6C9zfb7Mg1mbFKry/xzDAfmtLA=
+github.com/josharian/native v1.1.0/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2Cu7qoaN2w=
+github.com/jsimonetti/rtnetlink/v2 v2.0.2 h1:ZKlbCujrIpp4/u3V2Ka0oxlf4BCkt6ojkvpy3nZoCBY=
+github.com/jsimonetti/rtnetlink/v2 v2.0.2/go.mod h1:7MoNYNbb3UaDHtF8udiJo/RH6VsTKP1pqKLUTVCvToE=
+github.com/mdlayher/netlink v1.7.2 h1:/UtM3ofJap7Vl4QWCPDGXY8d3GIY2UGSDbK+QWmY8/g=
+github.com/mdlayher/netlink v1.7.2/go.mod h1:xraEF7uJbxLhc5fpHL4cPe221LI2bdttWlU+ZGLfQSw=
+github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U=
+github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA=
 github.com/opencontainers/selinux v1.11.0 h1:+5Zbo97w3Lbmb3PeqQtpmTkMwsW5nRI3YaLpt7tQ7oU=
 github.com/opencontainers/selinux v1.11.0/go.mod h1:E5dMC3VPuVvVHDYmi78qvhJp8+M586T4DlDRYpFkyec=
 github.com/peterbourgon/ff/v3 v3.4.0 h1:QBvM/rizZM1cB0p0lGMdmR7HxZeI/ZrBWB4DqLkMUBc=
@@ -8,6 +20,8 @@ go.uber.org/mock v0.4.0 h1:VcM4ZOtdbR4f6VXfiOpwpVJDL6lCReaZ6mw31wqh7KU=
 go.uber.org/mock v0.4.0/go.mod h1:a6FSlNadKUHUa9IP5Vyt1zh4fC7uAwxMutEAscFbkZc=
 golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw=
 golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U=
+golang.org/x/exp v0.0.0-20230224173230-c95f2b4c22f2 h1:Jvc7gsqn21cJHCmAWx0LiimpP18LZmUxkT5Mp7EZ1mI=
+golang.org/x/exp v0.0.0-20230224173230-c95f2b4c22f2/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/net v0.30.0 h1:AcW1SDZMkb8IpzCdQUaIq2sP4sZ4zw+55h6ynffypl4=
 golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU=
 golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=

internal/config/config.go

@@ -28,6 +28,8 @@ type Conf struct {
 	CustomSSHDPort              int
 	CustomSSHDCfgFile           string
 	AuthorizedKeysCheckInterval time.Duration
+
+	ManageReservedIPv6 bool
 }
 
 // Init initializes the agent's configuration
@@ -40,9 +42,12 @@ func Init() *Conf {
 
 	fs.BoolVar(&cfg.UseSyslog, "syslog", false, "Use syslog service for logging")
 	fs.BoolVar(&cfg.DebugMode, "debug", false, "Turn on debug mode")
+
 	fs.IntVar(&cfg.CustomSSHDPort, "sshd_port", 0, "The port sshd is binding to")
 	fs.StringVar(&cfg.CustomSSHDCfgFile, "sshd_config", "", "The location of sshd_config")
 
+	fs.BoolVar(&cfg.ManageReservedIPv6, "reserved_ipv6", false, "enable reserved IPv6 assignment/unassignment feature")
+
 	ff.Parse(fs, os.Args[1:],
 		ff.WithEnvVarPrefix("DROPLET_AGENT"),
 	)

internal/config/version.go

@@ -3,4 +3,4 @@
 package config
 
 // Version is the current package version.
-const Version = "v1.2.9"
+const Version = "v1.3.0"

internal/metadata/actioner/reserved_ipv6_actioner.go

@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package actioner
+
+import (
+	"sync/atomic"
+
+	"github.com/digitalocean/droplet-agent/internal/log"
+	"github.com/digitalocean/droplet-agent/internal/metadata"
+	"github.com/digitalocean/droplet-agent/internal/reservedipv6"
+)
+
+const (
+	logPrefix string = "[Reserved IPv6 Actioner]"
+)
+
+// NewReservedIPv6Actioner returns a new DigitalOcean Reserved IPv6 actioner
+func NewReservedIPv6Actioner(mgr reservedipv6.Manager) MetadataActioner {
+	return &reservedIPv6Actioner{
+		mgr:           mgr,
+		activeActions: &atomic.Uint32{},
+		closing:       &atomic.Bool{},
+		allDone:       make(chan struct{}, 1),
+	}
+}
+
+type reservedIPv6Actioner struct {
+	mgr           reservedipv6.Manager
+	activeActions *atomic.Uint32
+	closing       *atomic.Bool
+	allDone       chan struct{}
+}
+
+func (da *reservedIPv6Actioner) Do(md *metadata.Metadata) {
+	da.activeActions.Add(1)
+	defer func() {
+		// decrement active counter, then check shutdown state
+		ret := da.activeActions.Add(^uint32(0))
+		if ret == 0 && da.closing.Load() {
+			close(da.allDone)
+		}
+	}()
+
+	ipv6 := md.ReservedIP.IPv6
+
+	if ipv6.Active {
+		logDebug("Attempting to assign Reserved IPv6 address '%s'", ipv6.IPAddress)
+		if err := da.mgr.Assign(ipv6.IPAddress); err != nil {
+			logError("failed to assign Reserved IPv6 address '%s': %v", ipv6.IPAddress, err)
+			return
+		}
+		logInfo("Assigned Reserved IPv6 address '%s'", ipv6.IPAddress)
+	} else {
+		logDebug("Attempting to unassign all Reserved IPv6 addresses")
+		if err := da.mgr.Unassign(); err != nil {
+			logError("failed to unassign all Reserved IPv6 addresses: %v", err)
+			return
+		}
+		logInfo("Unassigned all Reserved IPv6 addresses")
+	}
+}
+
+func (da *reservedIPv6Actioner) Shutdown() {
+	logInfo("Shutting down")
+	da.closing.Store(true)
+
+	// if there are still jobs in progress, wait for them to finish
+	if da.activeActions.Load() > 0 {
+		logDebug("Waiting for jobs in progress")
+		<-da.allDone
+	}
+	logInfo("Bye-bye")
+}
+
+// logInfo wraps log.Info with rip6LogPrefix
+func logInfo(format string, params ...any) {
+	msg := logPrefix + " " + format
+	log.Info(msg, params)
+}
+
+// logDebug wraps log.Debug with rip6LogPrefix
+func logDebug(format string, params ...any) {
+	msg := logPrefix + " " + format
+	log.Debug(msg, params)
+}
+
+// logError wraps log.Error with rip6LogPrefix
+func logError(format string, params ...any) {
+	msg := logPrefix + " " + format
+	log.Error(msg, params)
+}

internal/metadata/actioner/reserved_ipv6_actioner_test.go

@@ -0,0 +1,54 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package actioner
+
+import (
+	"testing"
+
+	"github.com/digitalocean/droplet-agent/internal/metadata"
+	"github.com/digitalocean/droplet-agent/internal/reservedipv6"
+	"go.uber.org/mock/gomock"
+)
+
+func TestReservedIPv6Actioner_Do(t *testing.T) {
+	t.Run("assign", func(t *testing.T) {
+		// Arrange
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		rip6Manager := reservedipv6.NewMockManager(ctrl)
+		actioner := NewReservedIPv6Actioner(rip6Manager)
+
+		rip6Manager.EXPECT().Assign("2001:4860:4860::8888").Return(nil)
+
+		// Act
+		actioner.Do(&metadata.Metadata{
+			ReservedIP: &metadata.ReservedIP{
+				IPv6: &metadata.ReservedIPv6{
+					Active:    true,
+					IPAddress: "2001:4860:4860::8888",
+				},
+			},
+		})
+	})
+
+	t.Run("unassign", func(t *testing.T) {
+		// Arrange
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		rip6Manager := reservedipv6.NewMockManager(ctrl)
+		actioner := NewReservedIPv6Actioner(rip6Manager)
+
+		rip6Manager.EXPECT().Unassign().Return(nil)
+
+		// Act
+		actioner.Do(&metadata.Metadata{
+			ReservedIP: &metadata.ReservedIP{
+				IPv6: &metadata.ReservedIPv6{
+					Active: false,
+				},
+			},
+		})
+	})
+}

internal/metadata/metadata.go

@@ -30,6 +30,20 @@ type Metadata struct {
 	DOTTYStatus        AgentStatus `json:"dotty_status,omitempty"`
 	SSHInfo            *SSHInfo    `json:"ssh_info,omitempty"`
 	ManagedKeysEnabled *bool       `json:"managed_keys_enabled,omitempty"`
+
+	ReservedIP *ReservedIP `json:"reserved_ip,omitempty"`
+}
+
+// ReservedIP defines the Metadata fields of a Reserved IP.
+
+type ReservedIP struct {
+	IPv6 *ReservedIPv6 `json:"ipv6,omitempty"`
+}
+
+// ReservedIP defines the Metadata fields of a Reserved IPv6.
+type ReservedIPv6 struct {
+	IPAddress string `json:"ip_address,omitempty"`
+	Active    bool   `json:"active,omitempty"`
 }
 
 // SSHInfo contains the information of the sshd service running on the droplet

internal/metadata/updater/updater_test.go

@@ -35,7 +35,8 @@ func Test_agentInfoUpdaterImpl_Update(t *testing.T) {
 					ExpectedRequest: newRequest(t, []byte("{\"dotty_status\":\"running\",\"ssh_info\":{\"port\":256}}")),
 				}
 
-				client.EXPECT().Do(reqMatcher).Return(&http.Response{StatusCode: 202}, nil)
+				client.EXPECT().Do(reqMatcher).Return(&http.Response{StatusCode: 202, Body: respBody}, nil)
+				respBody.EXPECT().Close()
 			},
 			false,
 		},

internal/metadata/watcher/web_watcher.go

@@ -9,10 +9,10 @@ import (
 	"sync"
 	"time"
 
-	"github.com/digitalocean/droplet-agent/internal/log"
+	"golang.org/x/time/rate"
 
+	"github.com/digitalocean/droplet-agent/internal/log"
 	"github.com/digitalocean/droplet-agent/internal/metadata/actioner"
-	"golang.org/x/time/rate"
 )
 
 type webBasedWatcher struct {