Skip to content

Commit

Permalink
[controller] Add a sds scaner and a cache (#47)
Browse files Browse the repository at this point in the history
Signed-off-by: Viktor Kramarenko <[email protected]>
  • Loading branch information
ViktorKram authored May 21, 2024
1 parent fe2c39b commit 9cb4133
Show file tree
Hide file tree
Showing 12 changed files with 412 additions and 13 deletions.
10 changes: 10 additions & 0 deletions images/agent/cmd/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
goruntime "runtime"
"sds-node-configurator/api/v1alpha1"
"sds-node-configurator/config"
"sds-node-configurator/pkg/cache"
"sds-node-configurator/pkg/controller"
"sds-node-configurator/pkg/kubutils"
"sds-node-configurator/pkg/logger"
Expand Down Expand Up @@ -109,6 +110,15 @@ func main() {
}
log.Info("[main] ReTag ends")

sdsCache := cache.New()

go func() {
if err = controller.RunScanner(*log, *cfgParams, sdsCache); err != nil {
log.Error(err, "[main] unable to run scanner")
os.Exit(1)
}
}()

if _, err = controller.RunBlockDeviceController(ctx, mgr, *cfgParams, *log, metrics); err != nil {
log.Error(err, "[main] unable to controller.RunBlockDeviceController")
os.Exit(1)
Expand Down
43 changes: 35 additions & 8 deletions images/agent/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,16 +23,19 @@ import (
"os/exec"
"sds-node-configurator/internal"
"sds-node-configurator/pkg/logger"

"strconv"
"strings"
"time"
)

const (
ScanInterval = "SCAN_INTERVAL"
NodeName = "NODE_NAME"
LogLevel = "LOG_LEVEL"
MetricsPort = "METRICS_PORT"
MachineID = "MACHINE_ID"
ScanInterval = "SCAN_INTERVAL"
NodeName = "NODE_NAME"
LogLevel = "LOG_LEVEL"
MetricsPort = "METRICS_PORT"
MachineID = "MACHINE_ID"
ThrottleInterval = "THROTTLER_INTERVAL"
)

type Options struct {
Expand All @@ -43,6 +46,7 @@ type Options struct {
BlockDeviceScanInterval time.Duration
VolumeGroupScanInterval time.Duration
LLVRequeInterval time.Duration
ThrottleInterval time.Duration
}

func NewConfig() (*Options, error) {
Expand Down Expand Up @@ -71,9 +75,32 @@ func NewConfig() (*Options, error) {
opts.MetricsPort = ":8080"
}

opts.BlockDeviceScanInterval = 5
opts.VolumeGroupScanInterval = 5
opts.LLVRequeInterval = 5
scanInt := os.Getenv(ScanInterval)
if scanInt == "" {
opts.BlockDeviceScanInterval = 5
opts.VolumeGroupScanInterval = 5
opts.LLVRequeInterval = 5
} else {
interval, err := strconv.Atoi(scanInt)
if err != nil {
return nil, fmt.Errorf("[NewConfig] unable to get %s, error: %w", ScanInterval, err)
}
opts.BlockDeviceScanInterval = time.Duration(interval)
opts.VolumeGroupScanInterval = time.Duration(interval)
opts.LLVRequeInterval = time.Duration(interval)
}

thrInt := os.Getenv(ThrottleInterval)
if thrInt == "" {
opts.ThrottleInterval = 3
} else {
interval, err := strconv.Atoi(scanInt)
if err != nil {
return nil, fmt.Errorf("[NewConfig] unable to get %s, error: %w", ThrottleInterval, err)
}

opts.ThrottleInterval = time.Duration(interval)
}

return &opts, nil
}
Expand Down
1 change: 1 addition & 0 deletions images/agent/go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ require (
github.com/google/go-cmp v0.6.0
github.com/onsi/ginkgo/v2 v2.14.0
github.com/onsi/gomega v1.30.0
github.com/pilebones/go-udev v0.9.0
github.com/prometheus/client_golang v1.18.0
github.com/stretchr/testify v1.8.4
k8s.io/api v0.29.4
Expand Down
2 changes: 2 additions & 0 deletions images/agent/go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,8 @@ github.com/onsi/ginkgo/v2 v2.14.0 h1:vSmGj2Z5YPb9JwCWT6z6ihcUvDhuXLc3sJiqd3jMKAY
github.com/onsi/ginkgo/v2 v2.14.0/go.mod h1:JkUdW7JkN0V6rFvsHcJ478egV3XH9NxpD27Hal/PhZw=
github.com/onsi/gomega v1.30.0 h1:hvMK7xYz4D3HapigLTeGdId/NcfQx1VHMJc60ew99+8=
github.com/onsi/gomega v1.30.0/go.mod h1:9sxs+SwGrKI0+PWe4Fxa9tFQQBG5xSsSbMXOI8PPpoQ=
github.com/pilebones/go-udev v0.9.0 h1:N1uEO/SxUwtIctc0WLU0t69JeBxIYEYnj8lT/Nabl9Q=
github.com/pilebones/go-udev v0.9.0/go.mod h1:T2eI2tUSK0hA2WS5QLjXJUfQkluZQu+18Cqvem3CaXI=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
Expand Down
87 changes: 87 additions & 0 deletions images/agent/pkg/cache/cache.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
package cache

import (
"fmt"
"sds-node-configurator/internal"
"sds-node-configurator/pkg/logger"
)

type Cache struct {
devices []internal.Device
pvs []internal.PVData
vgs []internal.VGData
lvs []internal.LVData
}

func New() Cache {
return Cache{}
}

func (c *Cache) StoreDevices(devices []internal.Device) {
c.devices = devices
}

func (c *Cache) GetDevices() []internal.Device {
dst := make([]internal.Device, len(c.devices))
copy(dst, c.devices)

return dst
}

func (c *Cache) StorePVs(pvs []internal.PVData) {
c.pvs = pvs
}

func (c *Cache) GetPVs() []internal.PVData {
dst := make([]internal.PVData, len(c.pvs))
copy(dst, c.pvs)

return dst
}

func (c *Cache) StoreVGs(vgs []internal.VGData) {
c.vgs = vgs
}

func (c *Cache) GetVGs() []internal.VGData {
dst := make([]internal.VGData, len(c.vgs))
copy(dst, c.vgs)

return dst
}

func (c *Cache) StoreLVs(lvs []internal.LVData) {
c.lvs = lvs
}

func (c *Cache) GetLVs() []internal.LVData {
dst := make([]internal.LVData, len(c.lvs))
copy(dst, c.lvs)

return dst
}

func (c *Cache) PrintTheCache(log logger.Logger) {
log.Cache("*****************CACHE BEGIN*****************")
log.Cache("[Devices BEGIN]")
for _, d := range c.devices {
log.Cache(fmt.Sprintf(" Device Name: %s, size: %s, fsType: %s, serial: %s, wwn: %s", d.Name, d.Size.String(), d.FSType, d.Serial, d.Wwn))
}
log.Cache("[Devices ENDS]")
log.Cache("[PVs BEGIN]")
for _, pv := range c.pvs {
log.Cache(fmt.Sprintf(" PV Name: %s, VG Name: %s, size: %s, vgTags: %s", pv.PVName, pv.VGName, pv.PVSize.String(), pv.VGTags))
}
log.Cache("[PVs ENDS]")
log.Cache("[VGs BEGIN]")
for _, vg := range c.vgs {
log.Cache(fmt.Sprintf(" VG Name: %s, size: %s, free: %s, vgTags: %s", vg.VGName, vg.VGSize.String(), vg.VGFree.String(), vg.VGTags))
}
log.Cache("[VGs ENDS]")
log.Cache("[LVs BEGIN]")
for _, lv := range c.lvs {
log.Cache(fmt.Sprintf(" LV Name: %s, VG name: %s, size: %s, tags: %s, attr: %s, pool: %s", lv.LVName, lv.VGName, lv.LVSize.String(), lv.LvTags, lv.LVAttr, lv.PoolLv))
}
log.Cache("[LVs ENDS]")
log.Cache("*****************CACHE ENDS*****************")
}
68 changes: 68 additions & 0 deletions images/agent/pkg/cache/cache_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
package cache

import (
"github.com/stretchr/testify/assert"
"sds-node-configurator/internal"
"testing"
)

func TestCache(t *testing.T) {
sdsCache := New()
devices := []internal.Device{
{
Name: "test-1",
},
{
Name: "test-2",
},
{
Name: "test-3",
},
}

pvs := []internal.PVData{
{
PVName: "pv-1",
},
{
PVName: "pv-2",
},
{
PVName: "pv-3",
},
}

vgs := []internal.VGData{
{
VGName: "vg-1",
},
{
VGName: "vg-2",
},
{
VGName: "vg-3",
},
}

lvs := []internal.LVData{
{
LVName: "lv-1",
},
{
LVName: "lv-2",
},
{
LVName: "lv-3",
},
}

sdsCache.StoreDevices(devices)
sdsCache.StorePVs(pvs)
sdsCache.StoreVGs(vgs)
sdsCache.StoreLVs(lvs)

assert.ElementsMatch(t, devices, sdsCache.GetDevices())
assert.ElementsMatch(t, pvs, sdsCache.GetPVs())
assert.ElementsMatch(t, vgs, sdsCache.GetVGs())
assert.ElementsMatch(t, lvs, sdsCache.GetLVs())
}
2 changes: 1 addition & 1 deletion images/agent/pkg/controller/lvm_volume_group_discover.go
Original file line number Diff line number Diff line change
Expand Up @@ -516,7 +516,7 @@ func sortPVIssuesByVG(log logger.Logger, pvs []internal.PVData) map[string][]str
}

if stdErr.Len() != 0 {
log.Error(fmt.Errorf(stdErr.String()), fmt.Sprintf(`[sortPVIssuesByVG] pvs command for pv "%s" has stderr: `, pv.PVName))
log.Error(fmt.Errorf(stdErr.String()), fmt.Sprintf(`[sortPVIssuesByVG] pvs command for pv "%s" has stderr: %s`, pv.PVName, stdErr.String()))
pvIssuesByVG[pv.VGName+pv.VGUuid] = append(pvIssuesByVG[pv.VGName+pv.VGUuid], stdErr.String())
stdErr.Reset()
}
Expand Down
Loading

0 comments on commit 9cb4133

Please sign in to comment.