Merge pull request #3 from koordinator-sh/add-res-control

yarn-operator: add yarn resoruce controller

.gitignore

@@ -4,6 +4,7 @@
 *.dll
 *.so
 *.dylib
+bin
 
 # Test binary, built with `go test -c`
 *.test

.licenseignore

@@ -0,0 +1,4 @@
+vendor
+pkg/yarn/apis
+pkg/yarn/client/ipc
+pkg/yarn/config

Makefile

@@ -0,0 +1,203 @@
+# Git
+GIT_VERSION ?= $(shell git describe --tags --always)
+GIT_BRANCH ?= $(shell git rev-parse --abbrev-ref HEAD)
+GIT_COMMIT_ID ?= $(shell git rev-parse --short HEAD)
+
+# Image URL to use all building/pushing image targets
+REG ?= ghcr.io
+REG_NS ?= koordinator-sh
+REG_USER ?= ""
+REG_PWD ?= ""
+
+YARN_COPILOT_IMG ?= "${REG}/${REG_NS}/yarn-copilot:${GIT_BRANCH}-${GIT_COMMIT_ID}"
+YARN_OPERATOR_IMG ?= "${REG}/${REG_NS}/yarn-operator:${GIT_BRANCH}-${GIT_COMMIT_ID}"
+
+# ENVTEST_K8S_VERSION refers to the version of kubebuilder assets to be downloaded by envtest binary.
+ENVTEST_K8S_VERSION = 1.22
+
+AGENT_MODE ?= hostMode
+# Set license header files.
+LICENSE_HEADER_GO ?= hack/boilerplate/boilerplate.go.txt
+
+PACKAGES ?= $(shell go list ./... | grep -vE 'vendor|test/e2e')
+
+# Get the currently used golang install path (in GOPATH/bin, unless GOBIN is set)
+ifeq (,$(shell go env GOBIN))
+GOBIN=$(shell go env GOPATH)/bin
+else
+GOBIN=$(shell go env GOBIN)
+endif
+
+# Setting SHELL to bash allows bash commands to be executed by recipes.
+# This is a requirement for 'setup-envtest.sh' in the test target.
+# Options are set to exit when a recipe line exits non-zero or a piped command fails.
+SHELL = /usr/bin/env bash -o pipefail
+.SHELLFLAGS = -ec
+
+.PHONY: all
+all: build
+
+##@ General
+
+# The help target prints out all targets with their descriptions organized
+# beneath their categories. The categories are represented by '##@' and the
+# target descriptions by '##'. The awk commands is responsible for reading the
+# entire set of makefiles included in this invocation, looking for lines of the
+# file as xyz: ## something, and then pretty-format the target and help. Then,
+# if there's a line with ##@ something, that gets pretty-printed as a category.
+# More info on the usage of ANSI control characters for terminal formatting:
+# https://en.wikipedia.org/wiki/ANSI_escape_code#SGR_parameters
+# More info on the awk command:
+# http://linuxcommand.org/lc3_adv_awk.php
+
+.PHONY: help
+help: ## Display this help.
+	@awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n  make \033[36m<target>\033[0m\n"} /^[a-zA-Z_0-9-]+:.*?##/ { printf "  \033[36m%-15s\033[0m %s\n", $$1, $$2 } /^##@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) } ' $(MAKEFILE_LIST)
+
+##@ Development
+
+.PHONY: manifests
+manifests: controller-gen ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
+	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="./..." output:crd:artifacts:config=config/crd/bases
+
+.PHONY: generate
+generate: controller-gen ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations.
+	$(CONTROLLER_GEN) object:headerFile="$(LICENSE_HEADER_GO)" paths="./api/..."
+
+.PHONY: fmt
+fmt: ## Run go fmt against code.
+	go fmt ./...
+
+.PHONY: vet
+vet: ## Run go vet against code.
+	go vet ./...
+
+.PHONY: lint
+lint: lint-go lint-license ## Lint all code.
+
+.PHONY: lint-go
+lint-go: golangci-lint ## Lint Go code.
+	$(GOLANGCI_LINT) run -v --timeout=10m
+
+.PHONY: lint-license
+lint-license:
+	@hack/update-license-header.sh
+
+.PHONY: test
+test: manifests generate fmt vet envtest ## Run tests.
+	@KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" agent_mode=$(AGENT_MODE) go test $(PACKAGES) -race -covermode atomic -coverprofile cover.out
+
+.PHONY: fast-test
+fast-test: envtest ## Run tests fast.
+	@KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" agent_mode=$(AGENT_MODE) go test $(PACKAGES) -race -covermode atomic -coverprofile cover.out
+
+##@ Build
+
+.PHONY: build
+build: build-yarn-copilot build-yarn-operator
+
+.PHONY: build-yarn-copilot
+build-yarn-copilot: ## Build yarn-copilot binary.
+	go build -o bin/yarn-copilot cmd/yarn-copilot/main.go
+
+.PHONY: build-yarn-operator
+build-yarn-operator: ## Build yarn-operator binary.
+	go build -o bin/yarn-operator cmd/yarn-operator/main.go
+
+.PHONY: docker-build
+docker-build: test docker-build-yarn-copilot docker-build-yarn-operator
+
+.PHONY: docker-build-yarn-copilot
+docker-build-yarn-copilot: ## Build docker image with the yarn-copilot.
+	docker build --pull -t ${YARN_COPILOT_IMG} -f docker/yarn-copilot.dockerfile .
+
+.PHONY: docker-build-yarn-operator
+docker-build-yarn-operator: ## Build docker image with the yarn-operator.
+	docker build --pull -t ${YARN_OPERATOR_IMG} -f docker/yarn-operator.dockerfile .
+
+.PHONY: docker-push
+docker-push: docker-push-yarn-copilot docker-push-yarn-operator
+
+.PHONY: docker-push-yarn-copilot
+docker-push-yarn-copilot: ## Push docker image with the yarn-copilot.
+ifneq ($(REG_USER), "")
+	docker login -u $(REG_USER) -p $(REG_PWD) ${REG}
+endif
+	docker push ${YARN_COPILOT_IMG}
+
+.PHONY: docker-push-yarn-operator
+docker-push-yarn-operator: ## Push docker image with the yarn-operator.
+ifneq ($(REG_USER), "")
+	docker login -u $(REG_USER) -p $(REG_PWD) ${REG}
+endif
+	docker push ${YARN_OPERATOR_IMG}
+
+##@ Deployment
+
+ifndef ignore-not-found
+  ignore-not-found = false
+endif
+
+.PHONY: install
+install: manifests kustomize ## Install CRDs into the K8s cluster specified in ~/.kube/config.
+	$(KUSTOMIZE) build config/crd | kubectl apply -f -
+
+.PHONY: uninstall
+uninstall: manifests kustomize ## Uninstall CRDs from the K8s cluster specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource not found errors during deletion.
+	$(KUSTOMIZE) build config/crd | kubectl delete --ignore-not-found=$(ignore-not-found) -f -
+
+.PHONY: deploy
+deploy: manifests kustomize ## Deploy controller to the K8s cluster specified in ~/.kube/config.
+	cd config/manager && $(KUSTOMIZE) edit set image manager=$(YARN_OPERATOR_IMG) yarn-copilot=$(YARN_COPILOT_IMG)
+	@hack/kustomize.sh $(KUSTOMIZE) | kubectl apply -f -
+
+.PHONY: undeploy
+undeploy: ## Undeploy controller from the K8s cluster specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource not found errors during deletion.
+	@hack/kustomize.sh $(KUSTOMIZE) | kubectl delete --ignore-not-found=$(ignore-not-found) -f -
+
+##@ Build Dependencies
+
+## Location to install dependencies to
+LOCALBIN ?= $(shell pwd)/bin
+$(LOCALBIN):
+	mkdir -p $(LOCALBIN)
+
+## Tool Binaries
+KUSTOMIZE ?= $(LOCALBIN)/kustomize
+CONTROLLER_GEN ?= $(LOCALBIN)/controller-gen
+ENVTEST ?= $(LOCALBIN)/setup-envtest
+GOLANGCI_LINT ?= $(LOCALBIN)/golangci-lint
+GINKGO ?= $(LOCALBIN)/ginkgo
+HACK_DIR ?= $(PWD)/hack
+
+## Tool Versions
+KUSTOMIZE_VERSION ?= v3.8.7
+CONTROLLER_TOOLS_VERSION ?= v0.9.0
+GOLANGCILINT_VERSION ?= v1.47.3
+GINKGO_VERSION ?= v1.16.4
+
+KUSTOMIZE_INSTALL_SCRIPT ?= "https://raw.githubusercontent.com/kubernetes-sigs/kustomize/master/hack/install_kustomize.sh"
+.PHONY: kustomize
+kustomize: $(KUSTOMIZE) ## Download kustomize locally if necessary.
+$(KUSTOMIZE): $(LOCALBIN)
+	curl -s $(KUSTOMIZE_INSTALL_SCRIPT) | bash -s -- $(subst v,,$(KUSTOMIZE_VERSION)) $(LOCALBIN)
+
+.PHONY: controller-gen
+controller-gen: $(CONTROLLER_GEN) ## Download controller-gen locally if necessary.
+$(CONTROLLER_GEN): $(LOCALBIN)
+	GOBIN=$(LOCALBIN) go install sigs.k8s.io/controller-tools/cmd/controller-gen@$(CONTROLLER_TOOLS_VERSION)
+
+.PHONY: envtest
+envtest: $(ENVTEST) ## Download envtest-setup locally if necessary.
+$(ENVTEST): $(LOCALBIN)
+	GOBIN=$(LOCALBIN) go install sigs.k8s.io/controller-runtime/tools/setup-envtest@latest
+
+.PHONY: golangci-lint
+golangci-lint: $(GOLANGCI_LINT) ## Download golangci-lint locally if necessary.
+$(GOLANGCI_LINT): $(LOCALBIN)
+	GOBIN=$(LOCALBIN) go install github.com/golangci/golangci-lint/cmd/golangci-lint@$(GOLANGCILINT_VERSION)
+
+.PHONY: ginkgo
+ginkgo: $(GINKGO) ## Download ginkgo locally if necessary.
+$(GINKGO): $(LOCALBIN)
+	GOBIN=$(LOCALBIN) go install github.com/onsi/ginkgo/ginkgo@$(GINKGO_VERSION)

README.md

@@ -11,4 +11,4 @@ rm_update_node_resource.go is an example go YARN rpc client of rm-admin: call up
 
 # Run rm_update_node_resource
 change the `host` and `port` to target node id
-$ HADOOP_CONF_DIR=conf go run examples/rm_update_node_resource.go
+$ HADOOP_CONF_DIR=conf go run pkg/yarn/client/examples/rm_update_node_resource.go

cmd/yarn-operator/main.go

@@ -0,0 +1,147 @@
+/*
+Copyright 2022 The Koordinator 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 main
+
+import (
+	"flag"
+	"math/rand"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/spf13/pflag"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/klog/v2"
+	"k8s.io/klog/v2/klogr"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/koordinator-sh/goyarn/cmd/yarn-operator/options"
+	utilclient "github.com/koordinator-sh/koordinator/pkg/util/client"
+	"github.com/koordinator-sh/koordinator/pkg/util/fieldindex"
+)
+
+var (
+	setupLog = ctrl.Log.WithName("setup")
+
+	restConfigQPS   = flag.Int("rest-config-qps", 30, "QPS of rest config.")
+	restConfigBurst = flag.Int("rest-config-burst", 50, "Burst of rest config.")
+)
+
+func main() {
+	var metricsAddr, pprofAddr string
+	var healthProbeAddr string
+	var enableLeaderElection, enablePprof bool
+	var leaderElectionNamespace string
+	var namespace string
+	var syncPeriodStr string
+	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
+	flag.StringVar(&healthProbeAddr, "health-probe-addr", ":8000", "The address the healthz/readyz endpoint binds to.")
+	flag.BoolVar(&enableLeaderElection, "enable-leader-election", true, "Whether you need to enable leader election.")
+	flag.StringVar(&leaderElectionNamespace, "leader-election-namespace", "koordinator-system",
+		"This determines the namespace in which the leader election configmap will be created, it will use in-cluster namespace if empty.")
+	flag.StringVar(&namespace, "namespace", "",
+		"Namespace if specified restricts the manager's cache to watch objects in the desired namespace. Defaults to all namespaces.")
+	flag.BoolVar(&enablePprof, "enable-pprof", true, "Enable pprof for controller manager.")
+	flag.StringVar(&pprofAddr, "pprof-addr", ":8090", "The address the pprof binds to.")
+	flag.StringVar(&syncPeriodStr, "sync-period", "", "Determines the minimum frequency at which watched resources are reconciled.")
+	opts := options.NewOptions()
+	opts.InitFlags(flag.CommandLine)
+	//sloconfig.InitFlags(flag.CommandLine)
+	//utilfeature.DefaultMutableFeatureGate.DefaultMutableFeatureGateAddFlag(pflag.CommandLine)
+	klog.InitFlags(nil)
+	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
+	pflag.Parse()
+	rand.Seed(time.Now().UnixNano())
+	ctrl.SetLogger(klogr.New())
+	// features.SetDefaultFeatureGates()
+
+	if enablePprof {
+		go func() {
+			if err := http.ListenAndServe(pprofAddr, nil); err != nil {
+				setupLog.Error(err, "unable to start pprof")
+			}
+		}()
+	}
+
+	cfg := ctrl.GetConfigOrDie()
+	setRestConfig(cfg)
+	cfg.UserAgent = "koordinator-yarn-operator"
+
+	setupLog.Info("new clientset registry")
+	//err := extclient.NewRegistry(cfg)
+	//if err != nil {
+	//	setupLog.Error(err, "unable to init koordinator clientset and informer")
+	//	os.Exit(1)
+	//}
+
+	var syncPeriod *time.Duration
+	if syncPeriodStr != "" {
+		d, err := time.ParseDuration(syncPeriodStr)
+		if err != nil {
+			setupLog.Error(err, "invalid sync period flag")
+		} else {
+			syncPeriod = &d
+		}
+	}
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
+		Scheme:                     options.Scheme,
+		MetricsBindAddress:         metricsAddr,
+		HealthProbeBindAddress:     healthProbeAddr,
+		LeaderElection:             enableLeaderElection,
+		LeaderElectionID:           "koordinator-yarn-operator",
+		LeaderElectionNamespace:    leaderElectionNamespace,
+		LeaderElectionResourceLock: resourcelock.ConfigMapsResourceLock,
+		Namespace:                  namespace,
+		SyncPeriod:                 syncPeriod,
+		NewClient:                  utilclient.NewClient,
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to start manager")
+		os.Exit(1)
+	}
+
+	setupLog.Info("register field index")
+	if err := fieldindex.RegisterFieldIndexes(mgr.GetCache()); err != nil {
+		setupLog.Error(err, "failed to register field index")
+		os.Exit(1)
+	}
+
+	if err := opts.ApplyTo(mgr); err != nil {
+		setupLog.Error(err, "unable to setup controllers")
+		os.Exit(1)
+	}
+
+	// +kubebuilder:scaffold:builder
+
+	ctx := ctrl.SetupSignalHandler()
+
+	setupLog.Info("starting manager")
+	if err := mgr.Start(ctx); err != nil {
+		setupLog.Error(err, "problem running manager")
+		os.Exit(1)
+	}
+}
+
+func setRestConfig(c *rest.Config) {
+	if *restConfigQPS > 0 {
+		c.QPS = float32(*restConfigQPS)
+	}
+	if *restConfigBurst > 0 {
+		c.Burst = *restConfigBurst
+	}
+}