mirrors/node-feature-discovery
mirrors/node-feature-discovery
1
0
Fork 0
mirror of https://github.com/kubernetes-sigs/node-feature-discovery.git synced 2025-04-15 00:36:23 +00:00
Code Activity

Move most of functionality in cmd/ to pkg/

Browse source 
Move most of the code under cmd/nfd-master and cmd/nfd-worker into new
packages pkg/nfd-master and pk/nfd-worker, respectively. Makes extending
unit tests to "main" functions easier.
This commit is contained in:
Markus Lehtonen 2019-02-08 21:43:54 +02:00
parent 92b0cd9834
commit 2de0a019a3
10 changed files with 1249 additions and 1027 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

252
cmd/nfd-master/main.go
View file

@ -17,129 +17,48 @@ limitations under the License.
package main
import (
"crypto/tls"
"crypto/x509"
"fmt"
"io/ioutil"
"log"
"net"
"os"
"regexp"
"sort"
"strconv"
"strings"
"github.com/docopt/docopt-go"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/peer"
"sigs.k8s.io/node-feature-discovery/pkg/apihelper"
pb "sigs.k8s.io/node-feature-discovery/pkg/labeler"
master "sigs.k8s.io/node-feature-discovery/pkg/nfd-master"
"sigs.k8s.io/node-feature-discovery/pkg/version"
)
const (
// ProgramName is the canonical name of this program
ProgramName = "nfd-master"
// Namespace for feature labels
labelNs = "feature.node.kubernetes.io/"
// Namespace for all NFD-related annotations
annotationNs = "nfd.node.kubernetes.io/"
)
// package loggers
var (
stdoutLogger = log.New(os.Stdout, "", log.LstdFlags)
stderrLogger = log.New(os.Stderr, "", log.LstdFlags)
nodeName = os.Getenv("NODE_NAME")
)
// Labels are a Kubernetes representation of discovered features.
type Labels map[string]string
// Annotations are used for NFD-related node metadata
type Annotations map[string]string
// Command line arguments
type Args struct {
caFile string
certFile string
keyFile string
labelWhiteList *regexp.Regexp
noPublish bool
port int
verifyNodeName bool
}
func main() {
// Assert that the version is known
if version.Get() == "undefined" {
stderrLogger.Fatalf("version not set! Set -ldflags \"-X sigs.k8s.io/node-feature-discovery/pkg/version.version=`git describe --tags --dirty --always`\" during build or run.")
log.Fatalf("version not set! Set -ldflags \"-X sigs.k8s.io/node-feature-discovery/pkg/version.version=`git describe --tags --dirty --always`\" during build or run.")
}
stdoutLogger.Printf("Node Feature Discovery Master %s", version.Get())
stdoutLogger.Printf("NodeName: '%s'", nodeName)
// Parse command-line arguments.
args, err := argsParse(nil)
if err != nil {
stderrLogger.Fatalf("failed to parse command line: %v", err)
log.Fatalf("failed to parse command line: %v", err)
}
helper := apihelper.APIHelpers(apihelper.K8sHelpers{AnnotationNs: annotationNs,
LabelNs: labelNs})
if !args.noPublish {
err := updateMasterNode(helper)
if err != nil {
stderrLogger.Fatalf("failed to update master node: %v", err)
}
}
// Create server listening for TCP connections
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", args.port))
// Get new NfdMaster instance
instance, err := master.NewNfdMaster(args)
if err != nil {
stderrLogger.Fatalf("failed to listen: %v", err)
log.Fatalf("Failed to initialize NfdMaster instance: %v", err)
}
serverOpts := []grpc.ServerOption{}
// Enable mutual TLS authentication if --cert-file, --key-file or --ca-file
// is defined
if args.certFile != "" || args.keyFile != "" || args.caFile != "" {
// Load cert for authenticating this server
cert, err := tls.LoadX509KeyPair(args.certFile, args.keyFile)
if err != nil {
stderrLogger.Fatalf("failed to load server certificate: %v", err)
}
// Load CA cert for client cert verification
caCert, err := ioutil.ReadFile(args.caFile)
if err != nil {
stderrLogger.Fatalf("failed to read root certificate file: %v", err)
}
caPool := x509.NewCertPool()
if ok := caPool.AppendCertsFromPEM(caCert); !ok {
stderrLogger.Fatalf("failed to add certificate from '%s'", args.caFile)
}
// Create TLS config
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
ClientCAs: caPool,
ClientAuth: tls.RequireAndVerifyClientCert,
}
serverOpts = append(serverOpts, grpc.Creds(credentials.NewTLS(tlsConfig)))
if err = instance.Run(); err != nil {
log.Fatalf("ERROR: %v", err)
}
grpcServer := grpc.NewServer(serverOpts...)
pb.RegisterLabelerServer(grpcServer, &labelerServer{args: args, apiHelper: helper})
stdoutLogger.Printf("gRPC server serving on port: %d", args.port)
grpcServer.Serve(lis)
}
// argsParse parses the command line arguments passed to the program.
// The argument argv is passed only for testing purposes.
func argsParse(argv []string) (Args, error) {
args := Args{}
func argsParse(argv []string) (master.Args, error) {
args := master.Args{}
usage := fmt.Sprintf(`%s.
Usage:
@ -177,156 +96,19 @@ func argsParse(argv []string) (Args, error) {
// Parse argument values as usable types.
var err error
args.caFile = arguments["--ca-file"].(string)
args.certFile = arguments["--cert-file"].(string)
args.keyFile = arguments["--key-file"].(string)
args.noPublish = arguments["--no-publish"].(bool)
args.port, err = strconv.Atoi(arguments["--port"].(string))
args.CaFile = arguments["--ca-file"].(string)
args.CertFile = arguments["--cert-file"].(string)
args.KeyFile = arguments["--key-file"].(string)
args.NoPublish = arguments["--no-publish"].(bool)
args.Port, err = strconv.Atoi(arguments["--port"].(string))
if err != nil {
return args, fmt.Errorf("invalid --port defined: %s", err)
}
args.labelWhiteList, err = regexp.Compile(arguments["--label-whitelist"].(string))
args.LabelWhiteList, err = regexp.Compile(arguments["--label-whitelist"].(string))
if err != nil {
return args, fmt.Errorf("error parsing whitelist regex (%s): %s", arguments["--label-whitelist"], err)
}
args.verifyNodeName = arguments["--verify-node-name"].(bool)
args.VerifyNodeName = arguments["--verify-node-name"].(bool)
// Check TLS related args
if args.certFile != "" || args.keyFile != "" || args.caFile != "" {
if args.certFile == "" {
return args, fmt.Errorf("--cert-file needs to be specified alongside --key-file and --ca-file")
}
if args.keyFile == "" {
return args, fmt.Errorf("--key-file needs to be specified alongside --cert-file and --ca-file")
}
if args.caFile == "" {
return args, fmt.Errorf("--ca-file needs to be specified alongside --cert-file and --key-file")
}
}
return args, nil
}
// Advertise NFD master information
func updateMasterNode(helper apihelper.APIHelpers) error {
cli, err := helper.GetClient()
if err != nil {
return err
}
node, err := helper.GetNode(cli, nodeName)
if err != nil {
return err
}
// Advertise NFD version as an annotation
helper.AddAnnotations(node, Annotations{"master.version": version.Get()})
err = helper.UpdateNode(cli, node)
if err != nil {
stderrLogger.Printf("can't update node: %s", err.Error())
return err
}
return nil
}
// Filter labels if whitelist has been defined
func filterFeatureLabels(labels *Labels, labelWhiteList *regexp.Regexp) {
for name := range *labels {
// Skip if label doesn't match labelWhiteList
if !labelWhiteList.MatchString(name) {
stderrLogger.Printf("%s does not match the whitelist (%s) and will not be published.", name, labelWhiteList.String())
delete(*labels, name)
}
}
}
// Implement LabelerServer
type labelerServer struct {
args Args
apiHelper apihelper.APIHelpers
}
// Service SetLabels
func (s *labelerServer) SetLabels(c context.Context, r *pb.SetLabelsRequest) (*pb.SetLabelsReply, error) {
if s.args.verifyNodeName {
// Client authorization.
// Check that the node name matches the CN from the TLS cert
client, ok := peer.FromContext(c)
if !ok {
stderrLogger.Printf("gRPC request error: failed to get peer (client)")
return &pb.SetLabelsReply{}, fmt.Errorf("failed to get peer (client)")
}
tlsAuth, ok := client.AuthInfo.(credentials.TLSInfo)
if !ok {
stderrLogger.Printf("gRPC request error: incorrect client credentials from '%v'", client.Addr)
return &pb.SetLabelsReply{}, fmt.Errorf("incorrect client credentials")
}
if len(tlsAuth.State.VerifiedChains) == 0 || len(tlsAuth.State.VerifiedChains[0]) == 0 {
stderrLogger.Printf("gRPC request error: client certificate verification for '%v' failed", client.Addr)
return &pb.SetLabelsReply{}, fmt.Errorf("client certificate verification failed")
}
cn := tlsAuth.State.VerifiedChains[0][0].Subject.CommonName
if cn != r.NodeName {
stderrLogger.Printf("gRPC request error: authorization for %v failed: cert valid for '%s', requested node name '%s'", client.Addr, cn, r.NodeName)
return &pb.SetLabelsReply{}, fmt.Errorf("request authorization failed: cert valid for '%s', requested node name '%s'", cn, r.NodeName)
}
}
stdoutLogger.Printf("REQUEST Node: %s NFD-version: %s Labels: %s", r.NodeName, r.NfdVersion, r.Labels)
if !s.args.noPublish {
// Advertise NFD worker version and label names as annotations
keys := make([]string, 0, len(r.Labels))
for k, _ := range r.Labels {
keys = append(keys, k)
}
sort.Strings(keys)
annotations := Annotations{"worker.version": r.NfdVersion,
"feature-labels": strings.Join(keys, ",")}
err := updateNodeFeatures(s.apiHelper, r.NodeName, r.Labels, annotations)
if err != nil {
stderrLogger.Printf("failed to advertise labels: %s", err.Error())
return &pb.SetLabelsReply{}, err
}
}
return &pb.SetLabelsReply{}, nil
}
// advertiseFeatureLabels advertises the feature labels to a Kubernetes node
// via the API server.
func updateNodeFeatures(helper apihelper.APIHelpers, nodeName string, labels Labels, annotations Annotations) error {
cli, err := helper.GetClient()
if err != nil {
return err
}
// Get the worker node object
node, err := helper.GetNode(cli, nodeName)
if err != nil {
return err
}
// Remove old labels
if l, ok := node.Annotations[annotationNs+"feature-labels"]; ok {
oldLabels := strings.Split(l, ",")
helper.RemoveLabels(node, oldLabels)
}
// Also, remove all labels with the old prefix, and the old version label
helper.RemoveLabelsWithPrefix(node, "node.alpha.kubernetes-incubator.io/nfd")
helper.RemoveLabelsWithPrefix(node, "node.alpha.kubernetes-incubator.io/node-feature-discovery")
// Add labels to the node object.
helper.AddLabels(node, labels)
// Add annotations
helper.AddAnnotations(node, annotations)
// Send the updated node to the apiserver.
err = helper.UpdateNode(cli, node)
if err != nil {
stderrLogger.Printf("can't update node: %s", err.Error())
return err
}
return nil
}

215
cmd/nfd-master/main_test.go
View file

@ -17,206 +17,18 @@ limitations under the License.
package main
import (
"strings"
"testing"
. "github.com/smartystreets/goconvey/convey"
"github.com/stretchr/testify/mock"
"github.com/vektra/errors"
"golang.org/x/net/context"
api "k8s.io/api/core/v1"
k8sclient "k8s.io/client-go/kubernetes"
"sigs.k8s.io/node-feature-discovery/pkg/apihelper"
"sigs.k8s.io/node-feature-discovery/pkg/labeler"
"sigs.k8s.io/node-feature-discovery/pkg/version"
)
const (
mockNodeName = "mock-node"
)
func init() {
nodeName = mockNodeName
}
func TestUpdateNodeFeatures(t *testing.T) {
Convey("When I update the node using fake client", t, func() {
fakeFeatureLabels := map[string]string{"source-feature.1": "val1", "source-feature.2": "val2", "source-feature.3": "val3"}
fakeAnnotations := map[string]string{"version": version.Get()}
fakeFeatureLabelNames := make([]string, 0, len(fakeFeatureLabels))
for k, _ := range fakeFeatureLabels {
fakeFeatureLabelNames = append(fakeFeatureLabelNames, k)
}
fakeAnnotations["feature-labels"] = strings.Join(fakeFeatureLabelNames, ",")
mockAPIHelper := new(apihelper.MockAPIHelpers)
mockNode := &api.Node{}
mockClient := &k8sclient.Clientset{}
Convey("When I successfully update the node with feature labels", func() {
mockAPIHelper.On("GetClient").Return(mockClient, nil)
mockAPIHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil).Once()
mockAPIHelper.On("AddLabels", mockNode, fakeFeatureLabels).Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, labelNs).Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/nfd").Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/node-feature-discovery").Return().Once()
mockAPIHelper.On("AddAnnotations", mockNode, fakeAnnotations).Return().Once()
mockAPIHelper.On("UpdateNode", mockClient, mockNode).Return(nil).Once()
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is nil", func() {
So(err, ShouldBeNil)
})
})
Convey("When I fail to update the node with feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(nil, expectedError)
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
Convey("When I fail to get a mock client while updating feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(nil, expectedError)
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
Convey("When I fail to get a mock node while updating feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(mockClient, nil)
mockAPIHelper.On("GetNode", mockClient, mockNodeName).Return(nil, expectedError).Once()
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
Convey("When I fail to update a mock node while updating feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(mockClient, nil)
mockAPIHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil).Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, labelNs).Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/nfd").Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/node-feature-discovery").Return().Once()
mockAPIHelper.On("AddLabels", mockNode, fakeFeatureLabels).Return().Once()
mockAPIHelper.On("AddAnnotations", mockNode, fakeAnnotations).Return().Once()
mockAPIHelper.On("UpdateNode", mockClient, mockNode).Return(expectedError).Once()
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
})
}
func TestUpdateMasterNode(t *testing.T) {
Convey("When updating the nfd-master node", t, func() {
mockHelper := &apihelper.MockAPIHelpers{}
mockClient := &k8sclient.Clientset{}
mockNode := &api.Node{}
Convey("When update operation succeeds", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil)
mockHelper.On("AddAnnotations", mockNode, map[string]string{"master.version": version.Get()})
mockHelper.On("UpdateNode", mockClient, mockNode).Return(nil)
err := updateMasterNode(mockHelper)
Convey("No error should be returned", func() {
So(err, ShouldBeNil)
})
})
mockErr := errors.New("mock-error")
Convey("When getting API client fails", func() {
mockHelper.On("GetClient").Return(mockClient, mockErr)
err := updateMasterNode(mockHelper)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
Convey("When getting API node object fails", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, mockErr)
err := updateMasterNode(mockHelper)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
Convey("When updating node object fails", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil)
mockHelper.On("AddAnnotations", mock.Anything, mock.Anything)
mockHelper.On("UpdateNode", mockClient, mockNode).Return(mockErr)
err := updateMasterNode(mockHelper)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
})
}
func TestSetLabels(t *testing.T) {
Convey("When servicing SetLabels request", t, func() {
const workerName = "mock-worker"
const workerVer = "0.1-test"
mockHelper := &apihelper.MockAPIHelpers{}
mockClient := &k8sclient.Clientset{}
mockNode := &api.Node{}
mockServer := labelerServer{args: Args{}, apiHelper: mockHelper}
mockCtx := context.Background()
mockReq := &labeler.SetLabelsRequest{NodeName: workerName, NfdVersion: workerVer, Labels: map[string]string{"feature-1": "val-1"}}
Convey("When node update succeeds", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, workerName).Return(mockNode, nil)
mockHelper.On("RemoveLabelsWithPrefix", mockNode, mock.Anything).Return()
mockHelper.On("AddLabels", mockNode, mock.Anything).Return()
mockHelper.On("AddAnnotations", mockNode, map[string]string{"worker.version": workerVer, "feature-labels": "feature-1"})
mockHelper.On("UpdateNode", mockClient, mockNode).Return(nil)
_, err := mockServer.SetLabels(mockCtx, mockReq)
Convey("No error should be returned", func() {
So(err, ShouldBeNil)
})
})
mockErr := errors.New("mock-error")
Convey("When node update fails", func() {
mockHelper.On("GetClient").Return(mockClient, mockErr)
_, err := mockServer.SetLabels(mockCtx, mockReq)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
mockServer.args.noPublish = true
Convey("With '--no-publish'", func() {
_, err := mockServer.SetLabels(mockCtx, mockReq)
Convey("Operation should succeed", func() {
So(err, ShouldBeNil)
})
})
})
}
func TestArgsParse(t *testing.T) {
Convey("When parsing command line arguments", t, func() {
Convey("When --no-publish and --oneshot flags are passed", func() {
args, err := argsParse([]string{"--no-publish"})
Convey("noPublish is set and args.sources is set to the default value", func() {
So(args.noPublish, ShouldBeTrue)
So(len(args.labelWhiteList.String()), ShouldEqual, 0)
So(args.NoPublish, ShouldBeTrue)
So(len(args.LabelWhiteList.String()), ShouldEqual, 0)
So(err, ShouldBeNil)
})
})
@ -224,12 +36,12 @@ func TestArgsParse(t *testing.T) {
Convey("When valid args are specified", func() {
args, err := argsParse([]string{"--label-whitelist=.*rdt.*", "--port=1234", "--cert-file=crt", "--key-file=key", "--ca-file=ca"})
Convey("Argument parsing should succeed and args set to correct values", func() {
So(args.noPublish, ShouldBeFalse)
So(args.port, ShouldEqual, 1234)
So(args.certFile, ShouldEqual, "crt")
So(args.keyFile, ShouldEqual, "key")
So(args.caFile, ShouldEqual, "ca")
So(args.labelWhiteList.String(), ShouldResemble, ".*rdt.*")
So(args.NoPublish, ShouldBeFalse)
So(args.Port, ShouldEqual, 1234)
So(args.CertFile, ShouldEqual, "crt")
So(args.KeyFile, ShouldEqual, "key")
So(args.CaFile, ShouldEqual, "ca")
So(args.LabelWhiteList.String(), ShouldResemble, ".*rdt.*")
So(err, ShouldBeNil)
})
})
@ -239,16 +51,5 @@ func TestArgsParse(t *testing.T) {
So(err, ShouldNotBeNil)
})
})
Convey("When one of --cert-file, --key-file or --ca-file is missing", func() {
_, err := argsParse([]string{"--cert-file=crt", "--key-file=key"})
_, err2 := argsParse([]string{"--key-file=key", "--ca-file=ca"})
_, err3 := argsParse([]string{"--cert-file=crt", "--ca-file=ca"})
Convey("argsParse should fail", func() {
So(err, ShouldNotBeNil)
So(err2, ShouldNotBeNil)
So(err3, ShouldNotBeNil)
})
})
})
}

355
cmd/nfd-worker/main.go
View file

@ -17,39 +17,14 @@ limitations under the License.
package main
import (
"crypto/tls"
"crypto/x509"
"fmt"
"io/ioutil"
"log"
"os"
"regexp"
"strings"
"time"
"github.com/docopt/docopt-go"
"github.com/ghodss/yaml"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"k8s.io/apimachinery/pkg/util/validation"
pb "sigs.k8s.io/node-feature-discovery/pkg/labeler"
worker "sigs.k8s.io/node-feature-discovery/pkg/nfd-worker"
"sigs.k8s.io/node-feature-discovery/pkg/version"
"sigs.k8s.io/node-feature-discovery/source"
"sigs.k8s.io/node-feature-discovery/source/cpu"
"sigs.k8s.io/node-feature-discovery/source/cpuid"
"sigs.k8s.io/node-feature-discovery/source/fake"
"sigs.k8s.io/node-feature-discovery/source/iommu"
"sigs.k8s.io/node-feature-discovery/source/kernel"
"sigs.k8s.io/node-feature-discovery/source/local"
"sigs.k8s.io/node-feature-discovery/source/memory"
"sigs.k8s.io/node-feature-discovery/source/network"
"sigs.k8s.io/node-feature-discovery/source/panic_fake"
"sigs.k8s.io/node-feature-discovery/source/pci"
"sigs.k8s.io/node-feature-discovery/source/pstate"
"sigs.k8s.io/node-feature-discovery/source/rdt"
"sigs.k8s.io/node-feature-discovery/source/storage"
"sigs.k8s.io/node-feature-discovery/source/system"
)
const (
@ -57,135 +32,33 @@ const (
ProgramName = "nfd-worker"
)
// package loggers
var (
stdoutLogger = log.New(os.Stdout, "", log.LstdFlags)
stderrLogger = log.New(os.Stderr, "", log.LstdFlags)
nodeName = os.Getenv("NODE_NAME")
)
// Global config
type NFDConfig struct {
Sources struct {
Kernel *kernel.NFDConfig `json:"kernel,omitempty"`
Pci *pci.NFDConfig `json:"pci,omitempty"`
} `json:"sources,omitempty"`
}
var config = NFDConfig{}
// Labels are a Kubernetes representation of discovered features.
type Labels map[string]string
// Annotations are used for NFD-related node metadata
type Annotations map[string]string
// Command line arguments
type Args struct {
labelWhiteList string
caFile string
certFile string
keyFile string
configFile string
noPublish bool
options string
oneshot bool
server string
serverNameOverride string
sleepInterval time.Duration
sources []string
}
func main() {
// Assert that the version is known
if version.Get() == "undefined" {
stderrLogger.Fatalf("version not set! Set -ldflags \"-X sigs.k8s.io/node-feature-discovery/pkg/version.version=`git describe --tags --dirty --always`\" during build or run.")
log.Fatalf("version not set! Set -ldflags \"-X sigs.k8s.io/node-feature-discovery/pkg/version.version=`git describe --tags --dirty --always`\" during build or run.")
}
stdoutLogger.Printf("Node Feature Discovery Worker %s", version.Get())
stdoutLogger.Printf("NodeName: '%s'", nodeName)
// Parse command-line arguments.
args, err := argsParse(nil)
if err != nil {
stderrLogger.Fatalf("failed to parse command line: %v", err)
log.Fatalf("failed to parse command line: %v", err)
}
// Parse config
err = configParse(args.configFile, args.options)
// Get new NfdWorker instance
instance, err := worker.NewNfdWorker(args)
if err != nil {
stderrLogger.Print(err)
log.Fatalf("Failed to initialize NfdWorker instance: %v", err)
}
// Configure the parameters for feature discovery.
enabledSources, labelWhiteList, err := configureParameters(args.sources, args.labelWhiteList)
if err != nil {
stderrLogger.Fatalf("error occurred while configuring parameters: %s", err.Error())
}
// Connect to NFD server
dialOpts := []grpc.DialOption{}
if args.caFile != "" || args.certFile != "" || args.keyFile != "" {
// Load client cert for client authentication
cert, err := tls.LoadX509KeyPair(args.certFile, args.keyFile)
if err != nil {
stderrLogger.Fatalf("failed to load client certificate: %v", err)
}
// Load CA cert for server cert verification
caCert, err := ioutil.ReadFile(args.caFile)
if err != nil {
stderrLogger.Fatalf("failed to read root certificate file: %v", err)
}
caPool := x509.NewCertPool()
if ok := caPool.AppendCertsFromPEM(caCert); !ok {
stderrLogger.Fatalf("failed to add certificate from '%s'", args.caFile)
}
// Create TLS config
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: caPool,
ServerName: args.serverNameOverride,
}
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
} else {
dialOpts = append(dialOpts, grpc.WithInsecure())
}
conn, err := grpc.Dial(args.server, dialOpts...)
if err != nil {
stderrLogger.Fatalf("failed to connect: %v", err)
}
defer conn.Close()
client := pb.NewLabelerClient(conn)
for {
// Get the set of feature labels.
labels := createFeatureLabels(enabledSources, labelWhiteList)
// Update the node with the feature labels.
if !args.noPublish {
err := advertiseFeatureLabels(client, labels)
if err != nil {
stderrLogger.Fatalf("failed to advertise labels: %s", err.Error())
}
}
if args.oneshot {
break
}
if args.sleepInterval > 0 {
time.Sleep(args.sleepInterval)
} else {
conn.Close()
// Sleep forever
select {}
}
if err = instance.Run(); err != nil {
log.Fatalf("ERROR: %v", err)
}
}
// argsParse parses the command line arguments passed to the program.
// The argument argv is passed only for testing purposes.
func argsParse(argv []string) (Args, error) {
args := Args{}
func argsParse(argv []string) (worker.Args, error) {
args := worker.Args{}
usage := fmt.Sprintf(`%s.
Usage:
@ -238,204 +111,20 @@ func argsParse(argv []string) (Args, error) {
// Parse argument values as usable types.
var err error
args.caFile = arguments["--ca-file"].(string)
args.certFile = arguments["--cert-file"].(string)
args.configFile = arguments["--config"].(string)
args.keyFile = arguments["--key-file"].(string)
args.noPublish = arguments["--no-publish"].(bool)
args.options = arguments["--options"].(string)
args.server = arguments["--server"].(string)
args.serverNameOverride = arguments["--server-name-override"].(string)
args.sources = strings.Split(arguments["--sources"].(string), ",")
args.labelWhiteList = arguments["--label-whitelist"].(string)
args.oneshot = arguments["--oneshot"].(bool)
args.sleepInterval, err = time.ParseDuration(arguments["--sleep-interval"].(string))
// Check that sleep interval has a sane value
args.CaFile = arguments["--ca-file"].(string)
args.CertFile = arguments["--cert-file"].(string)
args.ConfigFile = arguments["--config"].(string)
args.KeyFile = arguments["--key-file"].(string)
args.NoPublish = arguments["--no-publish"].(bool)
args.Options = arguments["--options"].(string)
args.Server = arguments["--server"].(string)
args.ServerNameOverride = arguments["--server-name-override"].(string)
args.Sources = strings.Split(arguments["--sources"].(string), ",")
args.LabelWhiteList = arguments["--label-whitelist"].(string)
args.Oneshot = arguments["--oneshot"].(bool)
args.SleepInterval, err = time.ParseDuration(arguments["--sleep-interval"].(string))
if err != nil {
return args, fmt.Errorf("invalid --sleep-interval specified: %s", err.Error())
}
if args.sleepInterval > 0 && args.sleepInterval < time.Second {
stderrLogger.Printf("WARNING: too short sleep-intervall specified (%s), forcing to 1s", args.sleepInterval.String())
args.sleepInterval = time.Second
}
// Check TLS related args
if args.certFile != "" || args.keyFile != "" || args.caFile != "" {
if args.certFile == "" {
return args, fmt.Errorf("--cert-file needs to be specified alongside --key-file and --ca-file")
}
if args.keyFile == "" {
return args, fmt.Errorf("--key-file needs to be specified alongside --cert-file and --ca-file")
}
if args.caFile == "" {
return args, fmt.Errorf("--ca-file needs to be specified alongside --cert-file and --key-file")
}
}
return args, nil
}
// Parse configuration options
func configParse(filepath string, overrides string) error {
config.Sources.Kernel = &kernel.Config
config.Sources.Pci = &pci.Config
data, err := ioutil.ReadFile(filepath)
if err != nil {
return fmt.Errorf("Failed to read config file: %s", err)
}
// Read config file
err = yaml.Unmarshal(data, &config)
if err != nil {
return fmt.Errorf("Failed to parse config file: %s", err)
}
// Parse config overrides
err = yaml.Unmarshal([]byte(overrides), &config)
if err != nil {
return fmt.Errorf("Failed to parse --options: %s", err)
}
return nil
}
// configureParameters returns all the variables required to perform feature
// discovery based on command line arguments.
func configureParameters(sourcesWhiteList []string, labelWhiteListStr string) (enabledSources []source.FeatureSource, labelWhiteList *regexp.Regexp, err error) {
// A map for lookup
sourcesWhiteListMap := map[string]struct{}{}
for _, s := range sourcesWhiteList {
sourcesWhiteListMap[strings.TrimSpace(s)] = struct{}{}
}
// Configure feature sources.
allSources := []source.FeatureSource{
cpu.Source{},
cpuid.Source{},
fake.Source{},
iommu.Source{},
kernel.Source{},
memory.Source{},
network.Source{},
panic_fake.Source{},
pci.Source{},
pstate.Source{},
rdt.Source{},
storage.Source{},
system.Source{},
// local needs to be the last source so that it is able to override
// labels from other sources
local.Source{},
}
enabledSources = []source.FeatureSource{}
for _, s := range allSources {
if _, enabled := sourcesWhiteListMap[s.Name()]; enabled {
enabledSources = append(enabledSources, s)
}
}
// Compile labelWhiteList regex
labelWhiteList, err = regexp.Compile(labelWhiteListStr)
if err != nil {
stderrLogger.Printf("error parsing whitelist regex (%s): %s", labelWhiteListStr, err)
return nil, nil, err
}
return enabledSources, labelWhiteList, nil
}
// createFeatureLabels returns the set of feature labels from the enabled
// sources and the whitelist argument.
func createFeatureLabels(sources []source.FeatureSource, labelWhiteList *regexp.Regexp) (labels Labels) {
labels = Labels{}
// Do feature discovery from all configured sources.
for _, source := range sources {
labelsFromSource, err := getFeatureLabels(source)
if err != nil {
stderrLogger.Printf("discovery failed for source [%s]: %s", source.Name(), err.Error())
stderrLogger.Printf("continuing ...")
continue
}
for name, value := range labelsFromSource {
// Log discovered feature.
stdoutLogger.Printf("%s = %s", name, value)
// Skip if label doesn't match labelWhiteList
if !labelWhiteList.Match([]byte(name)) {
stderrLogger.Printf("%s does not match the whitelist (%s) and will not be published.", name, labelWhiteList.String())
continue
}
labels[name] = value
}
}
return labels
}
// getFeatureLabels returns node labels for features discovered by the
// supplied source.
func getFeatureLabels(source source.FeatureSource) (labels Labels, err error) {
defer func() {
if r := recover(); r != nil {
stderrLogger.Printf("panic occurred during discovery of source [%s]: %v", source.Name(), r)
err = fmt.Errorf("%v", r)
}
}()
labels = Labels{}
features, err := source.Discover()
if err != nil {
return nil, err
}
for k, v := range features {
// Validate label name
prefix := source.Name() + "-"
switch source.(type) {
case local.Source:
// Do not prefix labels from the hooks
prefix = ""
}
label := prefix + k
// Validate label name. Use dummy namespace 'ns' because there is no
// function to validate just the name part
errs := validation.IsQualifiedName("ns/" + label)
if len(errs) > 0 {
stderrLogger.Printf("Ignoring invalid feature name '%s': %s", label, errs)
continue
}
value := fmt.Sprintf("%v", v)
// Validate label value
errs = validation.IsValidLabelValue(value)
if len(errs) > 0 {
stderrLogger.Printf("Ignoring invalid feature value %s=%s: %s", label, value, errs)
continue
}
labels[label] = value
}
return labels, nil
}
// advertiseFeatureLabels advertises the feature labels to a Kubernetes node
// via the NFD server.
func advertiseFeatureLabels(client pb.LabelerClient, labels Labels) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
stdoutLogger.Printf("Sendng labeling request nfd-master")
labelReq := pb.SetLabelsRequest{Labels: labels,
NfdVersion: version.Get(),
NodeName: nodeName}
_, err := client.SetLabels(ctx, &labelReq)
if err != nil {
stderrLogger.Printf("failed to set node labels: %v", err)
return err
}
return nil
}

271
cmd/nfd-worker/main_test.go
View file

@ -17,77 +17,23 @@ limitations under the License.
package main
import (
"fmt"
"io/ioutil"
"os"
"regexp"
"testing"
"time"
. "github.com/smartystreets/goconvey/convey"
"github.com/stretchr/testify/mock"
"github.com/vektra/errors"
"sigs.k8s.io/node-feature-discovery/pkg/labeler"
"sigs.k8s.io/node-feature-discovery/source"
"sigs.k8s.io/node-feature-discovery/source/fake"
"sigs.k8s.io/node-feature-discovery/source/panic_fake"
)
func TestDiscoveryWithMockSources(t *testing.T) {
Convey("When I discover features from fake source and update the node using fake client", t, func() {
mockFeatureSource := new(source.MockFeatureSource)
fakeFeatureSourceName := string("testSource")
fakeFeatureNames := []string{"testfeature1", "testfeature2", "testfeature3"}
fakeFeatures := source.Features{}
fakeFeatureLabels := Labels{}
fakeFeatureLabelNames := make([]string, 0, len(fakeFeatureNames))
for _, f := range fakeFeatureNames {
fakeFeatures[f] = true
labelName := fakeFeatureSourceName + "-" + f
fakeFeatureLabels[labelName] = "true"
fakeFeatureLabelNames = append(fakeFeatureLabelNames, labelName)
}
fakeFeatureSource := source.FeatureSource(mockFeatureSource)
Convey("When I successfully get the labels from the mock source", func() {
mockFeatureSource.On("Name").Return(fakeFeatureSourceName)
mockFeatureSource.On("Discover").Return(fakeFeatures, nil)
returnedLabels, err := getFeatureLabels(fakeFeatureSource)
Convey("Proper label is returned", func() {
So(returnedLabels, ShouldResemble, fakeFeatureLabels)
})
Convey("Error is nil", func() {
So(err, ShouldBeNil)
})
})
Convey("When I fail to get the labels from the mock source", func() {
expectedError := errors.New("fake error")
mockFeatureSource.On("Discover").Return(nil, expectedError)
returnedLabels, err := getFeatureLabels(fakeFeatureSource)
Convey("No label is returned", func() {
So(returnedLabels, ShouldBeNil)
})
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
})
}
func TestArgsParse(t *testing.T) {
Convey("When parsing command line arguments", t, func() {
Convey("When --no-publish and --oneshot flags are passed", func() {
args, err := argsParse([]string{"--no-publish", "--oneshot"})
Convey("noPublish is set and args.sources is set to the default value", func() {
So(args.sleepInterval, ShouldEqual, 60*time.Second)
So(args.noPublish, ShouldBeTrue)
So(args.oneshot, ShouldBeTrue)
So(args.sources, ShouldResemble, []string{"cpu", "cpuid", "iommu", "kernel", "local", "memory", "network", "pci", "pstate", "rdt", "storage", "system"})
So(len(args.labelWhiteList), ShouldEqual, 0)
So(args.SleepInterval, ShouldEqual, 60*time.Second)
So(args.NoPublish, ShouldBeTrue)
So(args.Oneshot, ShouldBeTrue)
So(args.Sources, ShouldResemble, []string{"cpu", "cpuid", "iommu", "kernel", "local", "memory", "network", "pci", "pstate", "rdt", "storage", "system"})
So(len(args.LabelWhiteList), ShouldEqual, 0)
So(err, ShouldBeNil)
})
})
@ -96,11 +42,11 @@ func TestArgsParse(t *testing.T) {
args, err := argsParse([]string{"--sources=fake1,fake2,fake3", "--sleep-interval=30s"})
Convey("args.sources is set to appropriate values", func() {
So(args.sleepInterval, ShouldEqual, 30*time.Second)
So(args.noPublish, ShouldBeFalse)
So(args.oneshot, ShouldBeFalse)
So(args.sources, ShouldResemble, []string{"fake1", "fake2", "fake3"})
So(len(args.labelWhiteList), ShouldEqual, 0)
So(args.SleepInterval, ShouldEqual, 30*time.Second)
So(args.NoPublish, ShouldBeFalse)
So(args.Oneshot, ShouldBeFalse)
So(args.Sources, ShouldResemble, []string{"fake1", "fake2", "fake3"})
So(len(args.LabelWhiteList), ShouldEqual, 0)
So(err, ShouldBeNil)
})
})
@ -109,9 +55,9 @@ func TestArgsParse(t *testing.T) {
args, err := argsParse([]string{"--label-whitelist=.*rdt.*"})
Convey("args.labelWhiteList is set to appropriate value and args.sources is set to default value", func() {
So(args.noPublish, ShouldBeFalse)
So(args.sources, ShouldResemble, []string{"cpu", "cpuid", "iommu", "kernel", "local", "memory", "network", "pci", "pstate", "rdt", "storage", "system"})
So(args.labelWhiteList, ShouldResemble, ".*rdt.*")
So(args.NoPublish, ShouldBeFalse)
So(args.Sources, ShouldResemble, []string{"cpu", "cpuid", "iommu", "kernel", "local", "memory", "network", "pci", "pstate", "rdt", "storage", "system"})
So(args.LabelWhiteList, ShouldResemble, ".*rdt.*")
So(err, ShouldBeNil)
})
})
@ -120,193 +66,14 @@ func TestArgsParse(t *testing.T) {
args, err := argsParse([]string{"--no-publish", "--sources=fake1,fake2,fake3", "--ca-file=ca", "--cert-file=crt", "--key-file=key"})
Convey("--no-publish is set and args.sources is set to appropriate values", func() {
So(args.noPublish, ShouldBeTrue)
So(args.caFile, ShouldEqual, "ca")
So(args.certFile, ShouldEqual, "crt")
So(args.keyFile, ShouldEqual, "key")
So(args.sources, ShouldResemble, []string{"fake1", "fake2", "fake3"})
So(len(args.labelWhiteList), ShouldEqual, 0)
So(args.NoPublish, ShouldBeTrue)
So(args.CaFile, ShouldEqual, "ca")
So(args.CertFile, ShouldEqual, "crt")
So(args.KeyFile, ShouldEqual, "key")
So(args.Sources, ShouldResemble, []string{"fake1", "fake2", "fake3"})
So(len(args.LabelWhiteList), ShouldEqual, 0)
So(err, ShouldBeNil)
})
})
Convey("When one of --cert-file, --key-file or --ca-file is missing", func() {
_, err := argsParse([]string{"--cert-file=crt", "--key-file=key"})
_, err2 := argsParse([]string{"--key-file=key", "--ca-file=ca"})
_, err3 := argsParse([]string{"--cert-file=crt", "--ca-file=ca"})
Convey("Argument parsing should fail", func() {
So(err, ShouldNotBeNil)
So(err2, ShouldNotBeNil)
So(err3, ShouldNotBeNil)
})
})
})
}
func TestConfigParse(t *testing.T) {
Convey("When parsing configuration file", t, func() {
Convey("When non-accessible file is given", func() {
err := configParse("non-existing-file", "")
Convey("Should return error", func() {
So(err, ShouldNotBeNil)
})
})
// Create a temporary config file
f, err := ioutil.TempFile("", "nfd-test-")
defer os.Remove(f.Name())
So(err, ShouldBeNil)
f.WriteString(`sources:
kernel:
configOpts:
- "DMI"
pci:
deviceClassWhitelist:
- "ff"`)
f.Close()
Convey("When proper config file is given", func() {
err := configParse(f.Name(), "")
Convey("Should return error", func() {
So(err, ShouldBeNil)
So(config.Sources.Kernel.ConfigOpts, ShouldResemble, []string{"DMI"})
So(config.Sources.Pci.DeviceClassWhitelist, ShouldResemble, []string{"ff"})
})
})
})
}
func TestConfigureParameters(t *testing.T) {
Convey("When configuring parameters for node feature discovery", t, func() {
Convey("When no sourcesWhiteList and labelWhiteListStr are passed", func() {
sourcesWhiteList := []string{}
labelWhiteListStr := ""
emptyRegexp, _ := regexp.Compile("")
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error should not be produced", func() {
So(err, ShouldBeNil)
})
Convey("No sourcesWhiteList or labelWhiteList are returned", func() {
So(len(enabledSources), ShouldEqual, 0)
So(labelWhiteList, ShouldResemble, emptyRegexp)
})
})
Convey("When sourcesWhiteList is passed", func() {
sourcesWhiteList := []string{"fake"}
labelWhiteListStr := ""
emptyRegexp, _ := regexp.Compile("")
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error should not be produced", func() {
So(err, ShouldBeNil)
})
Convey("Proper sourcesWhiteList are returned", func() {
So(len(enabledSources), ShouldEqual, 1)
So(enabledSources[0], ShouldHaveSameTypeAs, fake.Source{})
So(labelWhiteList, ShouldResemble, emptyRegexp)
})
})
Convey("When invalid labelWhiteListStr is passed", func() {
sourcesWhiteList := []string{""}
labelWhiteListStr := "*"
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error is produced", func() {
So(enabledSources, ShouldBeNil)
So(labelWhiteList, ShouldBeNil)
So(err, ShouldNotBeNil)
})
})
Convey("When valid labelWhiteListStr is passed", func() {
sourcesWhiteList := []string{""}
labelWhiteListStr := ".*rdt.*"
expectRegexp, err := regexp.Compile(".*rdt.*")
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error should not be produced", func() {
So(err, ShouldBeNil)
})
Convey("Proper labelWhiteList is returned", func() {
So(len(enabledSources), ShouldEqual, 0)
So(labelWhiteList, ShouldResemble, expectRegexp)
})
})
})
}
func TestCreateFeatureLabels(t *testing.T) {
Convey("When creating feature labels from the configured sources", t, func() {
Convey("When fake feature source is configured", func() {
emptyLabelWL, _ := regexp.Compile("")
fakeFeatureSource := source.FeatureSource(new(fake.Source))
sources := []source.FeatureSource{}
sources = append(sources, fakeFeatureSource)
labels := createFeatureLabels(sources, emptyLabelWL)
Convey("Proper fake labels are returned", func() {
So(len(labels), ShouldEqual, 3)
So(labels, ShouldContainKey, "fake-fakefeature1")
So(labels, ShouldContainKey, "fake-fakefeature2")
So(labels, ShouldContainKey, "fake-fakefeature3")
})
})
Convey("When fake feature source is configured with a whitelist that doesn't match", func() {
emptyLabelWL, _ := regexp.Compile(".*rdt.*")
fakeFeatureSource := source.FeatureSource(new(fake.Source))
sources := []source.FeatureSource{}
sources = append(sources, fakeFeatureSource)
labels := createFeatureLabels(sources, emptyLabelWL)
Convey("fake labels are not returned", func() {
So(len(labels), ShouldEqual, 0)
So(labels, ShouldNotContainKey, "fake-fakefeature1")
So(labels, ShouldNotContainKey, "fake-fakefeature2")
So(labels, ShouldNotContainKey, "fake-fakefeature3")
})
})
})
}
func TestGetFeatureLabels(t *testing.T) {
Convey("When I get feature labels and panic occurs during discovery of a feature source", t, func() {
fakePanicFeatureSource := source.FeatureSource(new(panic_fake.Source))
returnedLabels, err := getFeatureLabels(fakePanicFeatureSource)
Convey("No label is returned", func() {
So(len(returnedLabels), ShouldEqual, 0)
})
Convey("Error is produced and panic error is returned", func() {
So(err, ShouldResemble, fmt.Errorf("fake panic error"))
})
})
}
func TestAdvertiseFeatureLabels(t *testing.T) {
Convey("When advertising labels", t, func() {
mockClient := &labeler.MockLabelerClient{}
labels := map[string]string{"feature-1": "value-1"}
Convey("Correct labeling request is sent", func() {
mockClient.On("SetLabels", mock.AnythingOfType("*context.timerCtx"), mock.AnythingOfType("*labeler.SetLabelsRequest")).Return(&labeler.SetLabelsReply{}, nil)
err := advertiseFeatureLabels(mockClient, labels)
Convey("There should be no error", func() {
So(err, ShouldBeNil)
})
})
Convey("Labeling request fails", func() {
mockErr := errors.New("mock-error")
mockClient.On("SetLabels", mock.AnythingOfType("*context.timerCtx"), mock.AnythingOfType("*labeler.SetLabelsRequest")).Return(&labeler.SetLabelsReply{}, mockErr)
err := advertiseFeatureLabels(mockClient, labels)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
})
}

211
pkg/nfd-master/nfd-master-internal_test.go Normal file
View file

@ -0,0 +1,211 @@
/*
Copyright 2019 The Kubernetes 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 nfdmaster
import (
"strings"
"testing"
. "github.com/smartystreets/goconvey/convey"
"github.com/stretchr/testify/mock"
"github.com/vektra/errors"
"golang.org/x/net/context"
api "k8s.io/api/core/v1"
k8sclient "k8s.io/client-go/kubernetes"
"sigs.k8s.io/node-feature-discovery/pkg/apihelper"
"sigs.k8s.io/node-feature-discovery/pkg/labeler"
"sigs.k8s.io/node-feature-discovery/pkg/version"
)
const (
mockNodeName = "mock-node"
)
func init() {
nodeName = mockNodeName
}
func TestUpdateNodeFeatures(t *testing.T) {
Convey("When I update the node using fake client", t, func() {
fakeFeatureLabels := map[string]string{"source-feature.1": "val1", "source-feature.2": "val2", "source-feature.3": "val3"}
fakeAnnotations := map[string]string{"version": version.Get()}
fakeFeatureLabelNames := make([]string, 0, len(fakeFeatureLabels))
for k, _ := range fakeFeatureLabels {
fakeFeatureLabelNames = append(fakeFeatureLabelNames, k)
}
fakeAnnotations["feature-labels"] = strings.Join(fakeFeatureLabelNames, ",")
mockAPIHelper := new(apihelper.MockAPIHelpers)
mockNode := &api.Node{}
mockClient := &k8sclient.Clientset{}
Convey("When I successfully update the node with feature labels", func() {
mockAPIHelper.On("GetClient").Return(mockClient, nil)
mockAPIHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil).Once()
mockAPIHelper.On("AddLabels", mockNode, fakeFeatureLabels).Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, labelNs).Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/nfd").Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/node-feature-discovery").Return().Once()
mockAPIHelper.On("AddAnnotations", mockNode, fakeAnnotations).Return().Once()
mockAPIHelper.On("UpdateNode", mockClient, mockNode).Return(nil).Once()
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is nil", func() {
So(err, ShouldBeNil)
})
})
Convey("When I fail to update the node with feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(nil, expectedError)
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
Convey("When I fail to get a mock client while updating feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(nil, expectedError)
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
Convey("When I fail to get a mock node while updating feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(mockClient, nil)
mockAPIHelper.On("GetNode", mockClient, mockNodeName).Return(nil, expectedError).Once()
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
Convey("When I fail to update a mock node while updating feature labels", func() {
expectedError := errors.New("fake error")
mockAPIHelper.On("GetClient").Return(mockClient, nil)
mockAPIHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil).Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, labelNs).Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/nfd").Return().Once()
mockAPIHelper.On("RemoveLabelsWithPrefix", mockNode, "node.alpha.kubernetes-incubator.io/node-feature-discovery").Return().Once()
mockAPIHelper.On("AddLabels", mockNode, fakeFeatureLabels).Return().Once()
mockAPIHelper.On("AddAnnotations", mockNode, fakeAnnotations).Return().Once()
mockAPIHelper.On("UpdateNode", mockClient, mockNode).Return(expectedError).Once()
err := updateNodeFeatures(mockAPIHelper, mockNodeName, fakeFeatureLabels, fakeAnnotations)
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
})
}
func TestUpdateMasterNode(t *testing.T) {
Convey("When updating the nfd-master node", t, func() {
mockHelper := &apihelper.MockAPIHelpers{}
mockClient := &k8sclient.Clientset{}
mockNode := &api.Node{}
Convey("When update operation succeeds", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil)
mockHelper.On("AddAnnotations", mockNode, map[string]string{"master.version": version.Get()})
mockHelper.On("UpdateNode", mockClient, mockNode).Return(nil)
err := updateMasterNode(mockHelper)
Convey("No error should be returned", func() {
So(err, ShouldBeNil)
})
})
mockErr := errors.New("mock-error")
Convey("When getting API client fails", func() {
mockHelper.On("GetClient").Return(mockClient, mockErr)
err := updateMasterNode(mockHelper)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
Convey("When getting API node object fails", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, mockErr)
err := updateMasterNode(mockHelper)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
Convey("When updating node object fails", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, mockNodeName).Return(mockNode, nil)
mockHelper.On("AddAnnotations", mock.Anything, mock.Anything)
mockHelper.On("UpdateNode", mockClient, mockNode).Return(mockErr)
err := updateMasterNode(mockHelper)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
})
}
func TestSetLabels(t *testing.T) {
Convey("When servicing SetLabels request", t, func() {
const workerName = "mock-worker"
const workerVer = "0.1-test"
mockHelper := &apihelper.MockAPIHelpers{}
mockClient := &k8sclient.Clientset{}
mockNode := &api.Node{}
mockServer := labelerServer{args: Args{}, apiHelper: mockHelper}
mockCtx := context.Background()
mockReq := &labeler.SetLabelsRequest{NodeName: workerName, NfdVersion: workerVer, Labels: map[string]string{"feature-1": "val-1"}}
Convey("When node update succeeds", func() {
mockHelper.On("GetClient").Return(mockClient, nil)
mockHelper.On("GetNode", mockClient, workerName).Return(mockNode, nil)
mockHelper.On("RemoveLabelsWithPrefix", mockNode, mock.Anything).Return()
mockHelper.On("AddLabels", mockNode, mock.Anything).Return()
mockHelper.On("AddAnnotations", mockNode, map[string]string{"worker.version": workerVer, "feature-labels": "feature-1"})
mockHelper.On("UpdateNode", mockClient, mockNode).Return(nil)
_, err := mockServer.SetLabels(mockCtx, mockReq)
Convey("No error should be returned", func() {
So(err, ShouldBeNil)
})
})
mockErr := errors.New("mock-error")
Convey("When node update fails", func() {
mockHelper.On("GetClient").Return(mockClient, mockErr)
_, err := mockServer.SetLabels(mockCtx, mockReq)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
mockServer.args.NoPublish = true
Convey("With '--no-publish'", func() {
_, err := mockServer.SetLabels(mockCtx, mockReq)
Convey("Operation should succeed", func() {
So(err, ShouldBeNil)
})
})
})
}

285
pkg/nfd-master/nfd-master.go Normal file
View file

@ -0,0 +1,285 @@
/*
Copyright 2019 The Kubernetes 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 nfdmaster
import (
"crypto/tls"
"crypto/x509"
"fmt"
"io/ioutil"
"log"
"net"
"os"
"regexp"
"sort"
"strings"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/peer"
"sigs.k8s.io/node-feature-discovery/pkg/apihelper"
pb "sigs.k8s.io/node-feature-discovery/pkg/labeler"
"sigs.k8s.io/node-feature-discovery/pkg/version"
)
const (
// Namespace for feature labels
labelNs = "feature.node.kubernetes.io/"
// Namespace for all NFD-related annotations
annotationNs = "nfd.node.kubernetes.io/"
)
// package loggers
var (
stdoutLogger = log.New(os.Stdout, "", log.LstdFlags)
stderrLogger = log.New(os.Stderr, "", log.LstdFlags)
nodeName = os.Getenv("NODE_NAME")
)
// Labels are a Kubernetes representation of discovered features.
type Labels map[string]string
// Annotations are used for NFD-related node metadata
type Annotations map[string]string
// Command line arguments
type Args struct {
CaFile string
CertFile string
KeyFile string
LabelWhiteList *regexp.Regexp
NoPublish bool
Port int
VerifyNodeName bool
}
type NfdMaster interface {
Run() error
Stop()
}
type nfdMaster struct {
args Args
server *grpc.Server
}
// Create new NfdMaster server instance.
func NewNfdMaster(args Args) (*nfdMaster, error) {
nfd := &nfdMaster{args: args}
// Check TLS related args
if args.CertFile != "" || args.KeyFile != "" || args.CaFile != "" {
if args.CertFile == "" {
return nfd, fmt.Errorf("--cert-file needs to be specified alongside --key-file and --ca-file")
}
if args.KeyFile == "" {
return nfd, fmt.Errorf("--key-file needs to be specified alongside --cert-file and --ca-file")
}
if args.CaFile == "" {
return nfd, fmt.Errorf("--ca-file needs to be specified alongside --cert-file and --key-file")
}
}
return nfd, nil
}
// Run NfdMaster server. The method returns in case of fatal errors or if Stop()
// is called.
func (m *nfdMaster) Run() error {
stdoutLogger.Printf("Node Feature Discovery Master %s", version.Get())
stdoutLogger.Printf("NodeName: '%s'", nodeName)
// Initialize Kubernetes API helpers
helper := apihelper.APIHelpers(apihelper.K8sHelpers{AnnotationNs: annotationNs,
LabelNs: labelNs})
if !m.args.NoPublish {
err := updateMasterNode(helper)
if err != nil {
return fmt.Errorf("failed to update master node: %v", err)
}
}
// Create server listening for TCP connections
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", m.args.Port))
if err != nil {
return fmt.Errorf("failed to listen: %v", err)
}
serverOpts := []grpc.ServerOption{}
// Enable mutual TLS authentication if --cert-file, --key-file or --ca-file
// is defined
if m.args.CertFile != "" || m.args.KeyFile != "" || m.args.CaFile != "" {
// Load cert for authenticating this server
cert, err := tls.LoadX509KeyPair(m.args.CertFile, m.args.KeyFile)
if err != nil {
return fmt.Errorf("failed to load server certificate: %v", err)
}
// Load CA cert for client cert verification
caCert, err := ioutil.ReadFile(m.args.CaFile)
if err != nil {
return fmt.Errorf("failed to read root certificate file: %v", err)
}
caPool := x509.NewCertPool()
if ok := caPool.AppendCertsFromPEM(caCert); !ok {
return fmt.Errorf("failed to add certificate from '%s'", m.args.CaFile)
}
// Create TLS config
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
ClientCAs: caPool,
ClientAuth: tls.RequireAndVerifyClientCert,
}
serverOpts = append(serverOpts, grpc.Creds(credentials.NewTLS(tlsConfig)))
}
m.server = grpc.NewServer(serverOpts...)
pb.RegisterLabelerServer(m.server, &labelerServer{args: m.args, apiHelper: helper})
stdoutLogger.Printf("gRPC server serving on port: %d", m.args.Port)
return m.server.Serve(lis)
}
// Stop NfdMaster
func (m *nfdMaster) Stop() {
m.server.Stop()
}
// Advertise NFD master information
func updateMasterNode(helper apihelper.APIHelpers) error {
cli, err := helper.GetClient()
if err != nil {
return err
}
node, err := helper.GetNode(cli, nodeName)
if err != nil {
return err
}
// Advertise NFD version as an annotation
helper.AddAnnotations(node, Annotations{"master.version": version.Get()})
err = helper.UpdateNode(cli, node)
if err != nil {
stderrLogger.Printf("can't update node: %s", err.Error())
return err
}
return nil
}
// Filter labels if whitelist has been defined
func filterFeatureLabels(labels *Labels, labelWhiteList *regexp.Regexp) {
for name := range *labels {
// Skip if label doesn't match labelWhiteList
if !labelWhiteList.MatchString(name) {
stderrLogger.Printf("%s does not match the whitelist (%s) and will not be published.", name, labelWhiteList.String())
delete(*labels, name)
}
}
}
// Implement LabelerServer
type labelerServer struct {
args Args
apiHelper apihelper.APIHelpers
}
// Service SetLabels
func (s *labelerServer) SetLabels(c context.Context, r *pb.SetLabelsRequest) (*pb.SetLabelsReply, error) {
if s.args.VerifyNodeName {
// Client authorization.
// Check that the node name matches the CN from the TLS cert
client, ok := peer.FromContext(c)
if !ok {
stderrLogger.Printf("gRPC request error: failed to get peer (client)")
return &pb.SetLabelsReply{}, fmt.Errorf("failed to get peer (client)")
}
tlsAuth, ok := client.AuthInfo.(credentials.TLSInfo)
if !ok {
stderrLogger.Printf("gRPC request error: incorrect client credentials from '%v'", client.Addr)
return &pb.SetLabelsReply{}, fmt.Errorf("incorrect client credentials")
}
if len(tlsAuth.State.VerifiedChains) == 0 || len(tlsAuth.State.VerifiedChains[0]) == 0 {
stderrLogger.Printf("gRPC request error: client certificate verification for '%v' failed", client.Addr)
return &pb.SetLabelsReply{}, fmt.Errorf("client certificate verification failed")
}
cn := tlsAuth.State.VerifiedChains[0][0].Subject.CommonName
if cn != r.NodeName {
stderrLogger.Printf("gRPC request error: authorization for %v failed: cert valid for '%s', requested node name '%s'", client.Addr, cn, r.NodeName)
return &pb.SetLabelsReply{}, fmt.Errorf("request authorization failed: cert valid for '%s', requested node name '%s'", cn, r.NodeName)
}
}
stdoutLogger.Printf("REQUEST Node: %s NFD-version: %s Labels: %s", r.NodeName, r.NfdVersion, r.Labels)
if !s.args.NoPublish {
// Advertise NFD worker version and label names as annotations
keys := make([]string, 0, len(r.Labels))
for k, _ := range r.Labels {
keys = append(keys, k)
}
sort.Strings(keys)
annotations := Annotations{"worker.version": r.NfdVersion,
"feature-labels": strings.Join(keys, ",")}
err := updateNodeFeatures(s.apiHelper, r.NodeName, r.Labels, annotations)
if err != nil {
stderrLogger.Printf("failed to advertise labels: %s", err.Error())
return &pb.SetLabelsReply{}, err
}
}
return &pb.SetLabelsReply{}, nil
}
// advertiseFeatureLabels advertises the feature labels to a Kubernetes node
// via the API server.
func updateNodeFeatures(helper apihelper.APIHelpers, nodeName string, labels Labels, annotations Annotations) error {
cli, err := helper.GetClient()
if err != nil {
return err
}
// Get the worker node object
node, err := helper.GetNode(cli, nodeName)
if err != nil {
return err
}
// Remove old labels
if l, ok := node.Annotations[annotationNs+"feature-labels"]; ok {
oldLabels := strings.Split(l, ",")
helper.RemoveLabels(node, oldLabels)
}
// Also, remove all labels with the old prefix, and the old version label
helper.RemoveLabelsWithPrefix(node, "node.alpha.kubernetes-incubator.io/nfd")
helper.RemoveLabelsWithPrefix(node, "node.alpha.kubernetes-incubator.io/node-feature-discovery")
// Add labels to the node object.
helper.AddLabels(node, labels)
// Add annotations
helper.AddAnnotations(node, annotations)
// Send the updated node to the apiserver.
err = helper.UpdateNode(cli, node)
if err != nil {
stderrLogger.Printf("can't update node: %s", err.Error())
return err
}
return nil
}

39
pkg/nfd-master/nfd-master_test.go Normal file
View file

@ -0,0 +1,39 @@
/*
Copyright 2019 The Kubernetes 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 nfdmaster_test
import (
"testing"
. "github.com/smartystreets/goconvey/convey"
m "sigs.k8s.io/node-feature-discovery/pkg/nfd-master"
)
func TestNewNfdMaster(t *testing.T) {
Convey("When initializing new NfdMaster instance", t, func() {
Convey("When one of --cert-file, --key-file or --ca-file is missing", func() {
_, err := m.NewNfdMaster(m.Args{CertFile: "crt", KeyFile: "key"})
_, err2 := m.NewNfdMaster(m.Args{KeyFile: "key", CaFile: "ca"})
_, err3 := m.NewNfdMaster(m.Args{CertFile: "crt", CaFile: "ca"})
Convey("An error should be returned", func() {
So(err, ShouldNotBeNil)
So(err2, ShouldNotBeNil)
So(err3, ShouldNotBeNil)
})
})
})
}

245
pkg/nfd-worker/nfd-worker-internal_test.go Normal file
View file

@ -0,0 +1,245 @@
/*
Copyright 2019 The Kubernetes 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 nfdworker
import (
"fmt"
"io/ioutil"
"os"
"regexp"
"testing"
. "github.com/smartystreets/goconvey/convey"
"github.com/stretchr/testify/mock"
"github.com/vektra/errors"
"sigs.k8s.io/node-feature-discovery/pkg/labeler"
"sigs.k8s.io/node-feature-discovery/source"
"sigs.k8s.io/node-feature-discovery/source/fake"
"sigs.k8s.io/node-feature-discovery/source/panic_fake"
)
func TestDiscoveryWithMockSources(t *testing.T) {
Convey("When I discover features from fake source and update the node using fake client", t, func() {
mockFeatureSource := new(source.MockFeatureSource)
fakeFeatureSourceName := string("testSource")
fakeFeatureNames := []string{"testfeature1", "testfeature2", "testfeature3"}
fakeFeatures := source.Features{}
fakeFeatureLabels := Labels{}
fakeFeatureLabelNames := make([]string, 0, len(fakeFeatureNames))
for _, f := range fakeFeatureNames {
fakeFeatures[f] = true
labelName := fakeFeatureSourceName + "-" + f
fakeFeatureLabels[labelName] = "true"
fakeFeatureLabelNames = append(fakeFeatureLabelNames, labelName)
}
fakeFeatureSource := source.FeatureSource(mockFeatureSource)
Convey("When I successfully get the labels from the mock source", func() {
mockFeatureSource.On("Name").Return(fakeFeatureSourceName)
mockFeatureSource.On("Discover").Return(fakeFeatures, nil)
returnedLabels, err := getFeatureLabels(fakeFeatureSource)
Convey("Proper label is returned", func() {
So(returnedLabels, ShouldResemble, fakeFeatureLabels)
})
Convey("Error is nil", func() {
So(err, ShouldBeNil)
})
})
Convey("When I fail to get the labels from the mock source", func() {
expectedError := errors.New("fake error")
mockFeatureSource.On("Discover").Return(nil, expectedError)
returnedLabels, err := getFeatureLabels(fakeFeatureSource)
Convey("No label is returned", func() {
So(returnedLabels, ShouldBeNil)
})
Convey("Error is produced", func() {
So(err, ShouldEqual, expectedError)
})
})
})
}
func TestConfigParse(t *testing.T) {
Convey("When parsing configuration file", t, func() {
Convey("When non-accessible file is given", func() {
err := configParse("non-existing-file", "")
Convey("Should return error", func() {
So(err, ShouldNotBeNil)
})
})
// Create a temporary config file
f, err := ioutil.TempFile("", "nfd-test-")
defer os.Remove(f.Name())
So(err, ShouldBeNil)
f.WriteString(`sources:
kernel:
configOpts:
- "DMI"
pci:
deviceClassWhitelist:
- "ff"`)
f.Close()
Convey("When proper config file is given", func() {
err := configParse(f.Name(), "")
Convey("Should return error", func() {
So(err, ShouldBeNil)
So(config.Sources.Kernel.ConfigOpts, ShouldResemble, []string{"DMI"})
So(config.Sources.Pci.DeviceClassWhitelist, ShouldResemble, []string{"ff"})
})
})
})
}
func TestConfigureParameters(t *testing.T) {
Convey("When configuring parameters for node feature discovery", t, func() {
Convey("When no sourcesWhiteList and labelWhiteListStr are passed", func() {
sourcesWhiteList := []string{}
labelWhiteListStr := ""
emptyRegexp, _ := regexp.Compile("")
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error should not be produced", func() {
So(err, ShouldBeNil)
})
Convey("No sourcesWhiteList or labelWhiteList are returned", func() {
So(len(enabledSources), ShouldEqual, 0)
So(labelWhiteList, ShouldResemble, emptyRegexp)
})
})
Convey("When sourcesWhiteList is passed", func() {
sourcesWhiteList := []string{"fake"}
labelWhiteListStr := ""
emptyRegexp, _ := regexp.Compile("")
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error should not be produced", func() {
So(err, ShouldBeNil)
})
Convey("Proper sourcesWhiteList are returned", func() {
So(len(enabledSources), ShouldEqual, 1)
So(enabledSources[0], ShouldHaveSameTypeAs, fake.Source{})
So(labelWhiteList, ShouldResemble, emptyRegexp)
})
})
Convey("When invalid labelWhiteListStr is passed", func() {
sourcesWhiteList := []string{""}
labelWhiteListStr := "*"
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error is produced", func() {
So(enabledSources, ShouldBeNil)
So(labelWhiteList, ShouldBeNil)
So(err, ShouldNotBeNil)
})
})
Convey("When valid labelWhiteListStr is passed", func() {
sourcesWhiteList := []string{""}
labelWhiteListStr := ".*rdt.*"
expectRegexp, err := regexp.Compile(".*rdt.*")
enabledSources, labelWhiteList, err := configureParameters(sourcesWhiteList, labelWhiteListStr)
Convey("Error should not be produced", func() {
So(err, ShouldBeNil)
})
Convey("Proper labelWhiteList is returned", func() {
So(len(enabledSources), ShouldEqual, 0)
So(labelWhiteList, ShouldResemble, expectRegexp)
})
})
})
}
func TestCreateFeatureLabels(t *testing.T) {
Convey("When creating feature labels from the configured sources", t, func() {
Convey("When fake feature source is configured", func() {
emptyLabelWL, _ := regexp.Compile("")
fakeFeatureSource := source.FeatureSource(new(fake.Source))
sources := []source.FeatureSource{}
sources = append(sources, fakeFeatureSource)
labels := createFeatureLabels(sources, emptyLabelWL)
Convey("Proper fake labels are returned", func() {
So(len(labels), ShouldEqual, 3)
So(labels, ShouldContainKey, "fake-fakefeature1")
So(labels, ShouldContainKey, "fake-fakefeature2")
So(labels, ShouldContainKey, "fake-fakefeature3")
})
})
Convey("When fake feature source is configured with a whitelist that doesn't match", func() {
emptyLabelWL, _ := regexp.Compile(".*rdt.*")
fakeFeatureSource := source.FeatureSource(new(fake.Source))
sources := []source.FeatureSource{}
sources = append(sources, fakeFeatureSource)
labels := createFeatureLabels(sources, emptyLabelWL)
Convey("fake labels are not returned", func() {
So(len(labels), ShouldEqual, 0)
So(labels, ShouldNotContainKey, "fake-fakefeature1")
So(labels, ShouldNotContainKey, "fake-fakefeature2")
So(labels, ShouldNotContainKey, "fake-fakefeature3")
})
})
})
}
func TestGetFeatureLabels(t *testing.T) {
Convey("When I get feature labels and panic occurs during discovery of a feature source", t, func() {
fakePanicFeatureSource := source.FeatureSource(new(panic_fake.Source))
returnedLabels, err := getFeatureLabels(fakePanicFeatureSource)
Convey("No label is returned", func() {
So(len(returnedLabels), ShouldEqual, 0)
})
Convey("Error is produced and panic error is returned", func() {
So(err, ShouldResemble, fmt.Errorf("fake panic error"))
})
})
}
func TestAdvertiseFeatureLabels(t *testing.T) {
Convey("When advertising labels", t, func() {
mockClient := &labeler.MockLabelerClient{}
labels := map[string]string{"feature-1": "value-1"}
Convey("Correct labeling request is sent", func() {
mockClient.On("SetLabels", mock.AnythingOfType("*context.timerCtx"), mock.AnythingOfType("*labeler.SetLabelsRequest")).Return(&labeler.SetLabelsReply{}, nil)
err := advertiseFeatureLabels(mockClient, labels)
Convey("There should be no error", func() {
So(err, ShouldBeNil)
})
})
Convey("Labeling request fails", func() {
mockErr := errors.New("mock-error")
mockClient.On("SetLabels", mock.AnythingOfType("*context.timerCtx"), mock.AnythingOfType("*labeler.SetLabelsRequest")).Return(&labeler.SetLabelsReply{}, mockErr)
err := advertiseFeatureLabels(mockClient, labels)
Convey("An error should be returned", func() {
So(err, ShouldEqual, mockErr)
})
})
})
}

364
pkg/nfd-worker/nfd-worker.go Normal file
View file

@ -0,0 +1,364 @@
/*
Copyright 2019 The Kubernetes 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 nfdworker
import (
"crypto/tls"
"crypto/x509"
"fmt"
"io/ioutil"
"log"
"os"
"regexp"
"strings"
"time"
"github.com/ghodss/yaml"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"k8s.io/apimachinery/pkg/util/validation"
pb "sigs.k8s.io/node-feature-discovery/pkg/labeler"
"sigs.k8s.io/node-feature-discovery/pkg/version"
"sigs.k8s.io/node-feature-discovery/source"
"sigs.k8s.io/node-feature-discovery/source/cpu"
"sigs.k8s.io/node-feature-discovery/source/cpuid"
"sigs.k8s.io/node-feature-discovery/source/fake"
"sigs.k8s.io/node-feature-discovery/source/iommu"
"sigs.k8s.io/node-feature-discovery/source/kernel"
"sigs.k8s.io/node-feature-discovery/source/local"
"sigs.k8s.io/node-feature-discovery/source/memory"
"sigs.k8s.io/node-feature-discovery/source/network"
"sigs.k8s.io/node-feature-discovery/source/panic_fake"
"sigs.k8s.io/node-feature-discovery/source/pci"
"sigs.k8s.io/node-feature-discovery/source/pstate"
"sigs.k8s.io/node-feature-discovery/source/rdt"
"sigs.k8s.io/node-feature-discovery/source/storage"
"sigs.k8s.io/node-feature-discovery/source/system"
)
// package loggers
var (
stdoutLogger = log.New(os.Stdout, "", log.LstdFlags)
stderrLogger = log.New(os.Stderr, "", log.LstdFlags)
nodeName = os.Getenv("NODE_NAME")
)
// Global config
type NFDConfig struct {
Sources struct {
Kernel *kernel.NFDConfig `json:"kernel,omitempty"`
Pci *pci.NFDConfig `json:"pci,omitempty"`
} `json:"sources,omitempty"`
}
var config = NFDConfig{}
// Labels are a Kubernetes representation of discovered features.
type Labels map[string]string
// Command line arguments
type Args struct {
LabelWhiteList string
CaFile string
CertFile string
KeyFile string
ConfigFile string
NoPublish bool
Options string
Oneshot bool
Server string
ServerNameOverride string
SleepInterval time.Duration
Sources []string
}
type NfdWorker interface {
Run() error
}
type nfdWorker struct {
args Args
}
// Create new NfdWorker instance.
func NewNfdWorker(args Args) (*nfdWorker, error) {
nfd := &nfdWorker{args: args}
if args.SleepInterval > 0 && args.SleepInterval < time.Second {
stderrLogger.Printf("WARNING: too short sleep-intervall specified (%s), forcing to 1s", args.SleepInterval.String())
args.SleepInterval = time.Second
}
// Check TLS related args
if args.CertFile != "" || args.KeyFile != "" || args.CaFile != "" {
if args.CertFile == "" {
return nfd, fmt.Errorf("--cert-file needs to be specified alongside --key-file and --ca-file")
}
if args.KeyFile == "" {
return nfd, fmt.Errorf("--key-file needs to be specified alongside --cert-file and --ca-file")
}
if args.CaFile == "" {
return nfd, fmt.Errorf("--ca-file needs to be specified alongside --cert-file and --key-file")
}
}
return nfd, nil
}
// Run NfdWorker client. Returns if a fatal error is encountered, or, after
// one request if OneShot is set to 'true' in the worker args.
func (w *nfdWorker) Run() error {
stdoutLogger.Printf("Node Feature Discovery Worker %s", version.Get())
stdoutLogger.Printf("NodeName: '%s'", nodeName)
// Parse config
err := configParse(w.args.ConfigFile, w.args.Options)
if err != nil {
stderrLogger.Print(err)
}
// Configure the parameters for feature discovery.
enabledSources, labelWhiteList, err := configureParameters(w.args.Sources, w.args.LabelWhiteList)
if err != nil {
return fmt.Errorf("error occurred while configuring parameters: %s", err.Error())
}
// Connect to NFD server
dialOpts := []grpc.DialOption{}
if w.args.CaFile != "" || w.args.CertFile != "" || w.args.KeyFile != "" {
// Load client cert for client authentication
cert, err := tls.LoadX509KeyPair(w.args.CertFile, w.args.KeyFile)
if err != nil {
return fmt.Errorf("failed to load client certificate: %v", err)
}
// Load CA cert for server cert verification
caCert, err := ioutil.ReadFile(w.args.CaFile)
if err != nil {
return fmt.Errorf("failed to read root certificate file: %v", err)
}
caPool := x509.NewCertPool()
if ok := caPool.AppendCertsFromPEM(caCert); !ok {
return fmt.Errorf("failed to add certificate from '%s'", w.args.CaFile)
}
// Create TLS config
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: caPool,
ServerName: w.args.ServerNameOverride,
}
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
} else {
dialOpts = append(dialOpts, grpc.WithInsecure())
}
conn, err := grpc.Dial(w.args.Server, dialOpts...)
if err != nil {
return fmt.Errorf("failed to connect: %v", err)
}
defer conn.Close()
client := pb.NewLabelerClient(conn)
for {
// Get the set of feature labels.
labels := createFeatureLabels(enabledSources, labelWhiteList)
// Update the node with the feature labels.
if !w.args.NoPublish {
err := advertiseFeatureLabels(client, labels)
if err != nil {
return fmt.Errorf("failed to advertise labels: %s", err.Error())
}
}
if w.args.Oneshot {
break
}
if w.args.SleepInterval > 0 {
time.Sleep(w.args.SleepInterval)
} else {
conn.Close()
// Sleep forever
select {}
}
}
return nil
}
// Parse configuration options
func configParse(filepath string, overrides string) error {
config.Sources.Kernel = &kernel.Config
config.Sources.Pci = &pci.Config
data, err := ioutil.ReadFile(filepath)
if err != nil {
return fmt.Errorf("Failed to read config file: %s", err)
}
// Read config file
err = yaml.Unmarshal(data, &config)
if err != nil {
return fmt.Errorf("Failed to parse config file: %s", err)
}
// Parse config overrides
err = yaml.Unmarshal([]byte(overrides), &config)
if err != nil {
return fmt.Errorf("Failed to parse --options: %s", err)
}
return nil
}
// configureParameters returns all the variables required to perform feature
// discovery based on command line arguments.
func configureParameters(sourcesWhiteList []string, labelWhiteListStr string) (enabledSources []source.FeatureSource, labelWhiteList *regexp.Regexp, err error) {
// A map for lookup
sourcesWhiteListMap := map[string]struct{}{}
for _, s := range sourcesWhiteList {
sourcesWhiteListMap[strings.TrimSpace(s)] = struct{}{}
}
// Configure feature sources.
allSources := []source.FeatureSource{
cpu.Source{},
cpuid.Source{},
fake.Source{},
iommu.Source{},
kernel.Source{},
memory.Source{},
network.Source{},
panic_fake.Source{},
pci.Source{},
pstate.Source{},
rdt.Source{},
storage.Source{},
system.Source{},
// local needs to be the last source so that it is able to override
// labels from other sources
local.Source{},
}
enabledSources = []source.FeatureSource{}
for _, s := range allSources {
if _, enabled := sourcesWhiteListMap[s.Name()]; enabled {
enabledSources = append(enabledSources, s)
}
}
// Compile labelWhiteList regex
labelWhiteList, err = regexp.Compile(labelWhiteListStr)
if err != nil {
stderrLogger.Printf("error parsing whitelist regex (%s): %s", labelWhiteListStr, err)
return nil, nil, err
}
return enabledSources, labelWhiteList, nil
}
// createFeatureLabels returns the set of feature labels from the enabled
// sources and the whitelist argument.
func createFeatureLabels(sources []source.FeatureSource, labelWhiteList *regexp.Regexp) (labels Labels) {
labels = Labels{}
// Do feature discovery from all configured sources.
for _, source := range sources {
labelsFromSource, err := getFeatureLabels(source)
if err != nil {
stderrLogger.Printf("discovery failed for source [%s]: %s", source.Name(), err.Error())
stderrLogger.Printf("continuing ...")
continue
}
for name, value := range labelsFromSource {
// Log discovered feature.
stdoutLogger.Printf("%s = %s", name, value)
// Skip if label doesn't match labelWhiteList
if !labelWhiteList.Match([]byte(name)) {
stderrLogger.Printf("%s does not match the whitelist (%s) and will not be published.", name, labelWhiteList.String())
continue
}
labels[name] = value
}
}
return labels
}
// getFeatureLabels returns node labels for features discovered by the
// supplied source.
func getFeatureLabels(source source.FeatureSource) (labels Labels, err error) {
defer func() {
if r := recover(); r != nil {
stderrLogger.Printf("panic occurred during discovery of source [%s]: %v", source.Name(), r)
err = fmt.Errorf("%v", r)
}
}()
labels = Labels{}
features, err := source.Discover()
if err != nil {
return nil, err
}
for k, v := range features {
// Validate label name
prefix := source.Name() + "-"
switch source.(type) {
case local.Source:
// Do not prefix labels from the hooks
prefix = ""
}
label := prefix + k
// Validate label name. Use dummy namespace 'ns' because there is no
// function to validate just the name part
errs := validation.IsQualifiedName("ns/" + label)
if len(errs) > 0 {
stderrLogger.Printf("Ignoring invalid feature name '%s': %s", label, errs)
continue
}
value := fmt.Sprintf("%v", v)
// Validate label value
errs = validation.IsValidLabelValue(value)
if len(errs) > 0 {
stderrLogger.Printf("Ignoring invalid feature value %s=%s: %s", label, value, errs)
continue
}
labels[label] = value
}
return labels, nil
}
// advertiseFeatureLabels advertises the feature labels to a Kubernetes node
// via the NFD server.
func advertiseFeatureLabels(client pb.LabelerClient, labels Labels) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
stdoutLogger.Printf("Sendng labeling request nfd-master")
labelReq := pb.SetLabelsRequest{Labels: labels,
NfdVersion: version.Get(),
NodeName: nodeName}
_, err := client.SetLabels(ctx, &labelReq)
if err != nil {
stderrLogger.Printf("failed to set node labels: %v", err)
return err
}
return nil
}

39
pkg/nfd-worker/nfd-worker_test.go Normal file
View file

@ -0,0 +1,39 @@
/*
Copyright 2019 The Kubernetes 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 nfdworker_test
import (
"testing"
. "github.com/smartystreets/goconvey/convey"
w "sigs.k8s.io/node-feature-discovery/pkg/nfd-worker"
)
func TestNewNfdWorker(t *testing.T) {
Convey("When initializing new NfdWorker instance", t, func() {
Convey("When one of --cert-file, --key-file or --ca-file is missing", func() {
_, err := w.NewNfdWorker(w.Args{CertFile: "crt", KeyFile: "key"})
_, err2 := w.NewNfdWorker(w.Args{KeyFile: "key", CaFile: "ca"})
_, err3 := w.NewNfdWorker(w.Args{CertFile: "crt", CaFile: "ca"})
Convey("An error should be returned", func() {
So(err, ShouldNotBeNil)
So(err2, ShouldNotBeNil)
So(err3, ShouldNotBeNil)
})
})
})
}