<commit_before>/*
Copyright 2017 The Rook Authors. All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

// Package flexvolume to manage Kubernetes storage attach events.
package flexvolume

import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/rpc"
	"os"
	"path"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/coreos/pkg/capnslog"
	"github.com/rook/rook/pkg/clusterd"
	"github.com/rook/rook/pkg/operator/ceph/agent"
	"github.com/rook/rook/pkg/operator/k8sutil"
	"k8s.io/kubernetes/pkg/volume/flexvolume"
)

const (
	UnixSocketName           = ".rook.sock"
	FlexvolumeVendor         = "ceph.rook.io"
	FlexvolumeVendorLegacy   = "rook.io"
	FlexDriverName           = "rook"
	flexvolumeDriverFileName = "rookflex"
	flexMountPath            = "/flexmnt/%s~%s"
	usrBinDir                = "/usr/local/bin/"
	settingsFilename         = "flex.config"
)

var logger = capnslog.NewPackageLogger("github.com/rook/rook", "flexvolume")

// FlexvolumeServer start a unix domain socket server to interact with the flexvolume driver
type FlexvolumeServer struct {
	context    *clusterd.Context
	controller *Controller
	listeners  map[string]net.Listener
}

// NewFlexvolumeServer creates an Flexvolume server
func NewFlexvolumeServer(context *clusterd.Context, controller *Controller) *FlexvolumeServer {
	return &FlexvolumeServer{
		context:    context,
		controller: controller,
		listeners:  make(map[string]net.Listener),
	}
}

// Start configures the flexvolume driver on the host and starts the unix domain socket server to communicate with the driver
func (s *FlexvolumeServer) Start(driverVendor, driverName string) error {
	// first install the flexvolume driver
	// /usr/local/bin/rookflex
	driverFile := path.Join(usrBinDir, flexvolumeDriverFileName)
	// /flexmnt/rook.io~rook-system
	flexVolumeDriverDir := fmt.Sprintf(flexMountPath, driverVendor, driverName)

	err := configureFlexVolume(driverFile, flexVolumeDriverDir, driverName)
	if err != nil {
		return fmt.Errorf("unable to configure flexvolume %s: %v", flexVolumeDriverDir, err)
	}

	unixSocketFile := path.Join(flexVolumeDriverDir, UnixSocketName) // /flextmnt/rook.io~rook-system/.rook.sock
	if _, ok := s.listeners[unixSocketFile]; ok {
		logger.Infof("flex server already running at %s", unixSocketFile)
		return nil
	}

	// remove unix socket if it existed previously
	if _, err := os.Stat(unixSocketFile); !os.IsNotExist(err) {
		logger.Info("Deleting unix domain socket file.")
		os.Remove(unixSocketFile)
	}

	listener, err := net.Listen("unix", unixSocketFile)
	if err != nil {
		return fmt.Errorf("unable to listen at %s: %v", unixSocketFile, err)
	}
	s.listeners[unixSocketFile] = listener

	if err := os.Chmod(unixSocketFile, 0770); err != nil {
		return fmt.Errorf("unable to set file permission to unix socket %s: %v", unixSocketFile, err)
	}

	go rpc.Accept(listener)

	logger.Infof("Listening on unix socket for Kubernetes volume attach commands: %s", unixSocketFile)
	return nil
}

// StopAll Stop the unix domain socket server and deletes the socket file
func (s *FlexvolumeServer) StopAll() {
	logger.Infof("Stopping %d unix socket rpc server(s).", len(s.listeners))
	for unixSocketFile, listener := range s.listeners {
		if err := listener.Close(); err != nil {
			logger.Errorf("Failed to stop unix socket rpc server: %+v", err)
		}

		// closing the listener should remove the unix socket file. But lets try it remove it just in case.
		if _, err := os.Stat(unixSocketFile); !os.IsNotExist(err) {
			logger.Infof("Deleting unix domain socket file %s.", unixSocketFile)
			os.Remove(unixSocketFile)
		}
	}
	s.listeners = make(map[string]net.Listener)
}

// RookDriverName return the Kubernetes version appropriate Rook driver name
func RookDriverName(context *clusterd.Context) (string, error) {
	// the driver name needs to be the same as the namespace so that we can support multiple namespaces
	// without the drivers conflicting with each other
	return os.Getenv(k8sutil.PodNamespaceEnvVar), nil
}

// TouchFlexDrivers causes k8s to reload the flex volumes. Needed periodically due to a k8s race condition with flex driver loading.
func TouchFlexDrivers(vendor, driverName string) {
	filename := path.Join(fmt.Sprintf(flexMountPath, vendor, driverName), driverName)
	logger.Debugf("reloading flex drivers. touching %s", filename)

	currenttime := time.Now().Local()
	err := os.Chtimes(filename, currenttime, currenttime)
	if err != nil {
		logger.Warningf("failed to touch file %s", filename)
	}
}

// Encode the flex settings in json
func generateFlexSettings(enableSELinuxRelabeling, enableFSGroup bool) ([]byte, error) {
	status := flexvolume.DriverStatus{
		Status: flexvolume.StatusSuccess,
		Capabilities: &flexvolume.DriverCapabilities{
			Attach: false,
			// Required for any mount performed on a host running selinux
			SELinuxRelabel: enableSELinuxRelabeling,
			FSGroup:        enableFSGroup,
		},
	}
	result, err := json.Marshal(status)
	if err != nil {
		return nil, fmt.Errorf("Invalid flex settings. %+v", err)
	}
	return result, nil
}

// The flex settings must be loaded from a file next to the flex driver since there is context
// that can be used other than the directory where the flex driver is running.
// This method cannot write to stdout since it is running in the context of the kubelet
// which only expects the json settings to be output.
func LoadFlexSettings(directory string) []byte {
	// Load the settings from the expected config file, ensure they are valid settings, then return them in
	// a json string to the caller
	var status flexvolume.DriverStatus
	if output, err := ioutil.ReadFile(path.Join(directory, settingsFilename)); err == nil {
		if err := json.Unmarshal(output, &status); err == nil {
			if output, err = json.Marshal(status); err == nil {
				return output
			}
		}
	}<commit_msg>ceph: add metrics for flexvolume driver<commit_after>/*
Copyright 2017 The Rook Authors. All rights reserved.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

// Package flexvolume to manage Kubernetes storage attach events.
package flexvolume

import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/rpc"
	"os"
	"path"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/coreos/pkg/capnslog"
	"github.com/rook/rook/pkg/clusterd"
	"github.com/rook/rook/pkg/operator/ceph/agent"
	"github.com/rook/rook/pkg/operator/k8sutil"
	"k8s.io/kubernetes/pkg/volume/flexvolume"
)

const (
	UnixSocketName           = ".rook.sock"
	FlexvolumeVendor         = "ceph.rook.io"
	FlexvolumeVendorLegacy   = "rook.io"
	FlexDriverName           = "rook"
	flexvolumeDriverFileName = "rookflex"
	flexMountPath            = "/flexmnt/%s~%s"
	usrBinDir                = "/usr/local/bin/"
	settingsFilename         = "flex.config"
)

var logger = capnslog.NewPackageLogger("github.com/rook/rook", "flexvolume")

// FlexvolumeServer start a unix domain socket server to interact with the flexvolume driver
type FlexvolumeServer struct {
	context    *clusterd.Context
	controller *Controller
	listeners  map[string]net.Listener
}

// NewFlexvolumeServer creates an Flexvolume server
func NewFlexvolumeServer(context *clusterd.Context, controller *Controller) *FlexvolumeServer {
	return &FlexvolumeServer{
		context:    context,
		controller: controller,
		listeners:  make(map[string]net.Listener),
	}
}

// Start configures the flexvolume driver on the host and starts the unix domain socket server to communicate with the driver
func (s *FlexvolumeServer) Start(driverVendor, driverName string) error {
	// first install the flexvolume driver
	// /usr/local/bin/rookflex
	driverFile := path.Join(usrBinDir, flexvolumeDriverFileName)
	// /flexmnt/rook.io~rook-system
	flexVolumeDriverDir := fmt.Sprintf(flexMountPath, driverVendor, driverName)

	err := configureFlexVolume(driverFile, flexVolumeDriverDir, driverName)
	if err != nil {
		return fmt.Errorf("unable to configure flexvolume %s: %v", flexVolumeDriverDir, err)
	}

	unixSocketFile := path.Join(flexVolumeDriverDir, UnixSocketName) // /flextmnt/rook.io~rook-system/.rook.sock
	if _, ok := s.listeners[unixSocketFile]; ok {
		logger.Infof("flex server already running at %s", unixSocketFile)
		return nil
	}

	// remove unix socket if it existed previously
	if _, err := os.Stat(unixSocketFile); !os.IsNotExist(err) {
		logger.Info("Deleting unix domain socket file.")
		os.Remove(unixSocketFile)
	}

	listener, err := net.Listen("unix", unixSocketFile)
	if err != nil {
		return fmt.Errorf("unable to listen at %s: %v", unixSocketFile, err)
	}
	s.listeners[unixSocketFile] = listener

	if err := os.Chmod(unixSocketFile, 0770); err != nil {
		return fmt.Errorf("unable to set file permission to unix socket %s: %v", unixSocketFile, err)
	}

	go rpc.Accept(listener)

	logger.Infof("Listening on unix socket for Kubernetes volume attach commands: %s", unixSocketFile)
	return nil
}

// StopAll Stop the unix domain socket server and deletes the socket file
func (s *FlexvolumeServer) StopAll() {
	logger.Infof("Stopping %d unix socket rpc server(s).", len(s.listeners))
	for unixSocketFile, listener := range s.listeners {
		if err := listener.Close(); err != nil {
			logger.Errorf("Failed to stop unix socket rpc server: %+v", err)
		}

		// closing the listener should remove the unix socket file. But lets try it remove it just in case.
		if _, err := os.Stat(unixSocketFile); !os.IsNotExist(err) {
			logger.Infof("Deleting unix domain socket file %s.", unixSocketFile)
			os.Remove(unixSocketFile)
		}
	}
	s.listeners = make(map[string]net.Listener)
}

// RookDriverName return the Kubernetes version appropriate Rook driver name
func RookDriverName(context *clusterd.Context) (string, error) {
	// the driver name needs to be the same as the namespace so that we can support multiple namespaces
	// without the drivers conflicting with each other
	return os.Getenv(k8sutil.PodNamespaceEnvVar), nil
}

// TouchFlexDrivers causes k8s to reload the flex volumes. Needed periodically due to a k8s race condition with flex driver loading.
func TouchFlexDrivers(vendor, driverName string) {
	filename := path.Join(fmt.Sprintf(flexMountPath, vendor, driverName), driverName)
	logger.Debugf("reloading flex drivers. touching %s", filename)

	currenttime := time.Now().Local()
	err := os.Chtimes(filename, currenttime, currenttime)
	if err != nil {
		logger.Warningf("failed to touch file %s", filename)
	}
}

// Encode the flex settings in json
func generateFlexSettings(enableSELinuxRelabeling, enableFSGroup bool) ([]byte, error) {
	status := flexvolume.DriverStatus{
		Status: flexvolume.StatusSuccess,
		Capabilities: &flexvolume.DriverCapabilities{
			Attach: false,
			// Required for metrics
			SupportsMetrics: true,
			// Required for any mount performed on a host running selinux
			SELinuxRelabel: enableSELinuxRelabeling,
			FSGroup:        enableFSGroup,
		},
	}
	result, err := json.Marshal(status)
	if err != nil {
		return nil, fmt.Errorf("Invalid flex settings. %+v", err)
	}
	return result, nil
}

// The flex settings must be loaded from a file next to the flex driver since there is context
// that can be used other than the directory where the flex driver is running.
// This method cannot write to stdout since it is running in the context of the kubelet
// which only expects the json settings to be output.
func LoadFlexSettings(directory string) []byte {
	// Load the settings from the expected config file, ensure they are valid settings, then return them in
	// a json string to the caller
	var status flexvolume.DriverStatus
	if output, err := ioutil.ReadFile(path.Join(directory, settingsFilename)); err == nil {
		if err := json.Unmarshal(output, &status); err == nil {
			if output, err = json.Marshal(status); err == nil {
				return output
			}
		}
	}
{'fraction_non_alphanumeric': 0.0997723526179449, 'fraction_numerical': 0.002668969306852971, 'mean_word_length': 4.247116968698517, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2128441', 'n_tokens_mistral': 4095, 'n_tokens_neox': 3885, 'n_words': 1300}