<commit_before>/*
Copyright 2015 The Kubernetes 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 minion

import (
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/signal"
	"path"
	"strconv"
	"strings"
	"syscall"

	kubeletapp "k8s.io/kubernetes/cmd/kubelet/app"
	exservice "k8s.io/kubernetes/contrib/mesos/pkg/executor/service"
	"k8s.io/kubernetes/contrib/mesos/pkg/hyperkube"
	"k8s.io/kubernetes/contrib/mesos/pkg/minion/config"
	"k8s.io/kubernetes/contrib/mesos/pkg/minion/tasks"
	"k8s.io/kubernetes/pkg/api/resource"
	client "k8s.io/kubernetes/pkg/client/unversioned"

	log "github.com/golang/glog"
	"github.com/kardianos/osext"
	"github.com/spf13/pflag"
	"gopkg.in/natefinch/lumberjack.v2"
)

const (
	proxyLogFilename    = "proxy.log"
	executorLogFilename = "executor.log"
)

type MinionServer struct {
	// embed the executor server to be able to use its flags
	// TODO(sttts): get rid of this mixing of the minion and the executor server with a multiflags implementation for km
	KubeletExecutorServer *exservice.KubeletExecutorServer

	privateMountNS bool
	hks            hyperkube.Interface
	clientConfig   *client.Config
	kmBinary       string
	tasks          []*tasks.Task

	pathOverride        string // the PATH environment for the sub-processes
	cgroupPrefix        string // e.g. mesos
	cgroupRoot          string // the cgroupRoot that we pass to the kubelet-executor, depends on containPodResources
	mesosCgroup         string // discovered mesos cgroup root, e.g. /mesos/{container-id}
	containPodResources bool

	logMaxSize      resource.Quantity
	logMaxBackups   int
	logMaxAgeInDays int
	logVerbosity    int32 // see glog.Level

	runProxy                       bool
	proxyLogV                      int
	proxyBindall                   bool
	proxyMode                      string
	conntrackMax                   int
	conntrackTCPTimeoutEstablished int
}

// NewMinionServer creates the MinionServer struct with default values to be used by hyperkube
func NewMinionServer() *MinionServer {
	s := &MinionServer{
		KubeletExecutorServer: exservice.NewKubeletExecutorServer(),
		privateMountNS:        false, // disabled until Docker supports customization of the parent mount namespace
		cgroupPrefix:          config.DefaultCgroupPrefix,
		containPodResources:   true,
		logMaxSize:            config.DefaultLogMaxSize(),
		logMaxBackups:         config.DefaultLogMaxBackups,
		logMaxAgeInDays:       config.DefaultLogMaxAgeInDays,
		runProxy:              true,
		proxyMode:             "userspace", // upstream default is "iptables" post-v1.1
	}

	// cache this for later use
	binary, err := osext.Executable()
	if err != nil {
		log.Fatalf("failed to determine currently running executable: %v", err)
	}
	s.kmBinary = binary

	return s
}

// filterArgsByFlagSet returns a list of args which are parsed by the given flag set
// and another list with those which do not match
func filterArgsByFlagSet(args []string, flags *pflag.FlagSet) ([]string, []string) {
	matched := []string{}
	notMatched := []string{}
	for _, arg := range args {
		err := flags.Parse([]string{arg})
		if err != nil {
			notMatched = append(notMatched, arg)
		} else {
			matched = append(matched, arg)
		}
	}
	return matched, notMatched
}

func findMesosCgroup(prefix string) (cgroupPath string, containerID string) {
	// derive our cgroup from MESOS_DIRECTORY environment
	mesosDir := os.Getenv("MESOS_DIRECTORY")
	if mesosDir == "" {
		log.V(2).Infof("cannot derive executor's cgroup because MESOS_DIRECTORY is empty")
		return
	}

	containerID = path.Base(mesosDir)
	if containerID == "" {
		log.V(2).Infof("cannot derive executor's cgroup from MESOS_DIRECTORY=%q", mesosDir)
		return
	}

	cgroupPath = path.Join("/", prefix, containerID)
	return
}

func (ms *MinionServer) launchProxyServer() {
	bindAddress := "0.0.0.0"
	if !ms.proxyBindall {
		bindAddress = ms.KubeletExecutorServer.Address
	}
	args := []string{
		fmt.Sprintf("--bind-address=%s", bindAddress),
		fmt.Sprintf("--v=%d", ms.proxyLogV),
		"--logtostderr=true",
		"--resource-container=" + path.Join("/", ms.mesosCgroup, "kube-proxy"),
		"--proxy-mode=" + ms.proxyMode,
		"--conntrack-max=" + strconv.Itoa(ms.conntrackMax),
		"--conntrack-tcp-timeout-established=" + strconv.Itoa(ms.conntrackTCPTimeoutEstablished),
	}

	if ms.clientConfig.Host != "" {
		args = append(args, fmt.Sprintf("--master=%s", ms.clientConfig.Host))
	}
	if ms.KubeletExecutorServer.HostnameOverride != "" {
		args = append(args, fmt.Sprintf("--hostname-override=%s", ms.KubeletExecutorServer.HostnameOverride))
	}

	ms.launchHyperkubeServer(hyperkube.CommandProxy, args, proxyLogFilename)
}

// launchExecutorServer returns a chan that closes upon kubelet-executor death. since the kubelet-
// executor doesn't support failover right now, the right thing to do is to fail completely since all
// pods will be lost upon restart and we want mesos to recover the resources from them.
func (ms *MinionServer) launchExecutorServer(containerID string) <-chan struct{} {
	allArgs := os.Args[1:]

	// filter out minion flags, leaving those for the executor
	executorFlags := pflag.NewFlagSet("executor", pflag.ContinueOnError)
	executorFlags.SetOutput(ioutil.Discard)
	ms.AddExecutorFlags(executorFlags)
	executorArgs, _ := filterArgsByFlagSet(allArgs, executorFlags)

	executorArgs = append(executorArgs, "--resource-container="+path.Join("/", ms.mesosCgroup, "kubelet"))
	if ms.cgroupRoot != "" {
		executorArgs = append(executorArgs, "--cgroup-root="+ms.cgroupRoot)
	}

	// forward containerID so that the executor may pass it along to containers that it launches
	var ctidOpt tasks.Option
	ctidOpt = func(t *tasks.Task) tasks.Option {
		oldenv := t.Env[:]
		t.Env = append(t.Env, "MESOS_EXECUTOR_CONTAINER_UUID="+containerID)
		return func(t2 *tasks.Task) tasks.Option {
			t2.Env = oldenv
			return ctidOpt
		}
	}

	// run executor and quit minion server when this exits cleanly
	execDied := make(chan struct{})
	ms.launchHyperkubeServer(hyperkube.CommandExecutor, executorArgs, executorLogFilename, tasks.NoRespawn(execDied), ctidOpt)
	return execDied
}

func (ms *MinionServer) launchHyperkubeServer(server string, args []string, logFileName string, options ...tasks.Option) {
	log.V(2).Infof("Spawning hyperkube %v with args '%+v'", server, args)

	kmArgs := append([]string{server}, args...)
	maxSize := ms.logMaxSize.Value()
	if maxSize > 0 {
		// convert to MB<commit_msg>stop creating /kubelet and /kube-proxy sub-containers because mesos slaves cannot handle it upon failover<commit_after>/*
Copyright 2015 The Kubernetes 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 minion

import (
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/signal"
	"path"
	"strconv"
	"strings"
	"syscall"

	kubeletapp "k8s.io/kubernetes/cmd/kubelet/app"
	exservice "k8s.io/kubernetes/contrib/mesos/pkg/executor/service"
	"k8s.io/kubernetes/contrib/mesos/pkg/hyperkube"
	"k8s.io/kubernetes/contrib/mesos/pkg/minion/config"
	"k8s.io/kubernetes/contrib/mesos/pkg/minion/tasks"
	"k8s.io/kubernetes/pkg/api/resource"
	client "k8s.io/kubernetes/pkg/client/unversioned"

	log "github.com/golang/glog"
	"github.com/kardianos/osext"
	"github.com/spf13/pflag"
	"gopkg.in/natefinch/lumberjack.v2"
)

const (
	proxyLogFilename    = "proxy.log"
	executorLogFilename = "executor.log"
)

type MinionServer struct {
	// embed the executor server to be able to use its flags
	// TODO(sttts): get rid of this mixing of the minion and the executor server with a multiflags implementation for km
	KubeletExecutorServer *exservice.KubeletExecutorServer

	privateMountNS bool
	hks            hyperkube.Interface
	clientConfig   *client.Config
	kmBinary       string
	tasks          []*tasks.Task

	pathOverride        string // the PATH environment for the sub-processes
	cgroupPrefix        string // e.g. mesos
	cgroupRoot          string // the cgroupRoot that we pass to the kubelet-executor, depends on containPodResources
	mesosCgroup         string // discovered mesos cgroup root, e.g. /mesos/{container-id}
	containPodResources bool

	logMaxSize      resource.Quantity
	logMaxBackups   int
	logMaxAgeInDays int
	logVerbosity    int32 // see glog.Level

	runProxy                       bool
	proxyLogV                      int
	proxyBindall                   bool
	proxyMode                      string
	conntrackMax                   int
	conntrackTCPTimeoutEstablished int
}

// NewMinionServer creates the MinionServer struct with default values to be used by hyperkube
func NewMinionServer() *MinionServer {
	s := &MinionServer{
		KubeletExecutorServer: exservice.NewKubeletExecutorServer(),
		privateMountNS:        false, // disabled until Docker supports customization of the parent mount namespace
		cgroupPrefix:          config.DefaultCgroupPrefix,
		containPodResources:   true,
		logMaxSize:            config.DefaultLogMaxSize(),
		logMaxBackups:         config.DefaultLogMaxBackups,
		logMaxAgeInDays:       config.DefaultLogMaxAgeInDays,
		runProxy:              true,
		proxyMode:             "userspace", // upstream default is "iptables" post-v1.1
	}

	// cache this for later use
	binary, err := osext.Executable()
	if err != nil {
		log.Fatalf("failed to determine currently running executable: %v", err)
	}
	s.kmBinary = binary

	return s
}

// filterArgsByFlagSet returns a list of args which are parsed by the given flag set
// and another list with those which do not match
func filterArgsByFlagSet(args []string, flags *pflag.FlagSet) ([]string, []string) {
	matched := []string{}
	notMatched := []string{}
	for _, arg := range args {
		err := flags.Parse([]string{arg})
		if err != nil {
			notMatched = append(notMatched, arg)
		} else {
			matched = append(matched, arg)
		}
	}
	return matched, notMatched
}

func findMesosCgroup(prefix string) (cgroupPath string, containerID string) {
	// derive our cgroup from MESOS_DIRECTORY environment
	mesosDir := os.Getenv("MESOS_DIRECTORY")
	if mesosDir == "" {
		log.V(2).Infof("cannot derive executor's cgroup because MESOS_DIRECTORY is empty")
		return
	}

	containerID = path.Base(mesosDir)
	if containerID == "" {
		log.V(2).Infof("cannot derive executor's cgroup from MESOS_DIRECTORY=%q", mesosDir)
		return
	}

	cgroupPath = path.Join("/", prefix, containerID)
	return
}

func (ms *MinionServer) launchProxyServer() {
	bindAddress := "0.0.0.0"
	if !ms.proxyBindall {
		bindAddress = ms.KubeletExecutorServer.Address
	}
	args := []string{
		fmt.Sprintf("--bind-address=%s", bindAddress),
		fmt.Sprintf("--v=%d", ms.proxyLogV),
		"--logtostderr=true",
		// TODO(jdef) resource-container is going away completely at some point, but
		// we need to override it here to disable the current default behavior
		"--resource-container=", // disable this; mesos slave doesn't like sub-containers yet
		"--proxy-mode=" + ms.proxyMode,
		"--conntrack-max=" + strconv.Itoa(ms.conntrackMax),
		"--conntrack-tcp-timeout-established=" + strconv.Itoa(ms.conntrackTCPTimeoutEstablished),
	}

	if ms.clientConfig.Host != "" {
		args = append(args, fmt.Sprintf("--master=%s", ms.clientConfig.Host))
	}
	if ms.KubeletExecutorServer.HostnameOverride != "" {
		args = append(args, fmt.Sprintf("--hostname-override=%s", ms.KubeletExecutorServer.HostnameOverride))
	}

	ms.launchHyperkubeServer(hyperkube.CommandProxy, args, proxyLogFilename)
}

// launchExecutorServer returns a chan that closes upon kubelet-executor death. since the kubelet-
// executor doesn't support failover right now, the right thing to do is to fail completely since all
// pods will be lost upon restart and we want mesos to recover the resources from them.
func (ms *MinionServer) launchExecutorServer(containerID string) <-chan struct{} {
	allArgs := os.Args[1:]

	// filter out minion flags, leaving those for the executor
	executorFlags := pflag.NewFlagSet("executor", pflag.ContinueOnError)
	executorFlags.SetOutput(ioutil.Discard)
	ms.AddExecutorFlags(executorFlags)
	executorArgs, _ := filterArgsByFlagSet(allArgs, executorFlags)

	// disable resource-container; mesos slave doesn't like sub-containers yet
	executorArgs = append(executorArgs, "--kubelet-cgroups=")
	if ms.cgroupRoot != "" {
		executorArgs = append(executorArgs, "--cgroup-root="+ms.cgroupRoot)
	}

	// forward containerID so that the executor may pass it along to containers that it launches
	var ctidOpt tasks.Option
	ctidOpt = func(t *tasks.Task) tasks.Option {
		oldenv := t.Env[:]
		t.Env = append(t.Env, "MESOS_EXECUTOR_CONTAINER_UUID="+containerID)
		return func(t2 *tasks.Task) tasks.Option {
			t2.Env = oldenv
			return ctidOpt
		}
	}

	// run executor and quit minion server when this exits cleanly
	execDied := make(chan struct{})
	ms.launchHyperkubeServer(hyperkube.CommandExecutor, executorArgs, executorLogFilename, tasks.NoRespawn(execDied), ctidOpt)
	return execDied
}

func (ms *MinionServer) launchHyperkubeServer(server string, args []string, logFileName string, options ...tasks.Option) {
	log.V(2).Infof("Spawning hyperkube %v with args '%+v'", server, args)

	kmArgs := append([]string{server}, args...)
	maxSize := ms.logMaxSize.Value()
	if maxSize > 0 {
		// convert to MB
{'fraction_non_alphanumeric': 0.10568817737479634, 'fraction_numerical': 0.00467521428065453, 'mean_word_length': 3.8682758620689657, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, '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': '7371582', 'n_tokens_mistral': 4724, 'n_tokens_neox': 4480, 'n_words': 1244}