penguin/monok8s

clitools/pkg/controller/admission/server.go

raw · 3733 bytes

package admission

import (
	"context"
	"io"
	"net/http"

	admissionv1 "k8s.io/api/admission/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	monov1alpha1 "example.com/monok8s/pkg/apis/monok8s/v1alpha1"
	"example.com/monok8s/pkg/controller/osupgrade"
	"example.com/monok8s/pkg/kube"

	"github.com/emicklei/go-restful/v3"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/apiserver/pkg/server/httplog"
	"k8s.io/klog/v2"
)

var (
	scheme       = runtime.NewScheme()
	codecs       = serializer.NewCodecFactory(scheme)
	deserializer = codecs.UniversalDeserializer()
)

var statusesNoTracePred = httplog.StatusIsNot(
	http.StatusOK,
	http.StatusFound,
	http.StatusMovedPermanently,
	http.StatusTemporaryRedirect,
	http.StatusBadRequest,
	http.StatusNotFound,
	http.StatusSwitchingProtocols,
)

func init() {
	_ = admissionv1.AddToScheme(scheme)
	_ = monov1alpha1.AddToScheme(scheme)
}

type Server struct {
	restfulCont *restful.Container

	ctx       context.Context
	clients   *kube.Clients
	namespace string
	nodeName  string
}

func NewServer(ctx context.Context, clients *kube.Clients, namespace, nodeName string) *Server {
	s := &Server{
		ctx:       ctx,
		clients:   clients,
		namespace: namespace,
		nodeName:  nodeName,
	}
	s.Initialize()
	return s
}

func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	if s == nil {
		http.Error(w, "admission server is nil", http.StatusInternalServerError)
		return
	}
	if s.restfulCont == nil {
		http.Error(w, "admission server not initialized", http.StatusInternalServerError)
		return
	}

	handler := httplog.WithLogging(s.restfulCont, statusesNoTracePred)
	handler.ServeHTTP(w, req)
}

func (s *Server) Initialize() {
	s.restfulCont = restful.NewContainer()

	ws := new(restful.WebService)

	ws.Path("/admission").
		Consumes(restful.MIME_JSON).
		Produces(restful.MIME_JSON)

	ws.Route(ws.POST("").To(s.triggerAdmission).
		Reads(admissionv1.AdmissionReview{}).
		Writes(admissionv1.AdmissionReview{}))

	s.restfulCont.Add(ws)
}

func (s *Server) triggerAdmission(request *restful.Request, response *restful.Response) {
	body, err := io.ReadAll(request.Request.Body)
	if err != nil {
		_ = response.WriteError(http.StatusBadRequest, err)
		return
	}

	var reviewReq admissionv1.AdmissionReview
	if _, _, err := deserializer.Decode(body, nil, &reviewReq); err != nil {
		_ = response.WriteError(http.StatusBadRequest, err)
		return
	}
	if reviewReq.Request == nil {
		_ = response.WriteErrorString(http.StatusBadRequest, "missing admission request")
		return
	}

	resp := admissionv1.AdmissionReview{
		TypeMeta: metav1.TypeMeta{
			Kind:       "AdmissionReview",
			APIVersion: "admission.k8s.io/v1",
		},
		Response: &admissionv1.AdmissionResponse{
			UID:     reviewReq.Request.UID,
			Allowed: true,
			Result:  &metav1.Status{Message: "OK"},
		},
	}

	var osu monov1alpha1.OSUpgrade
	if _, _, err := deserializer.Decode(reviewReq.Request.Object.Raw, nil, &osu); err != nil {
		klog.V(1).InfoS("Skipping non-OSUpgrade resource",
			"uid", reviewReq.Request.UID,
			"kind", reviewReq.Request.Kind.Kind,
			"operation", reviewReq.Request.Operation,
			"err", err,
		)
		_ = response.WriteEntity(resp)
		return
	}

	klog.InfoS("Received OSUpgrade admission",
		"uid", reviewReq.Request.UID,
		"operation", reviewReq.Request.Operation,
		"name", osu.Name,
		"namespace", osu.Namespace,
		"node", s.nodeName,
	)

	// Resolve every node name
	if err := osupgrade.EnsureOSUpgradeProgressForNode(
		s.ctx,
		s.clients,
		s.namespace,
		s.nodeName,
		&osu,
	); err != nil {
		klog.ErrorS(err, "ensure OSUpgradeProgress for node failed",
			"osupgrade", osu.Name,
			"node", s.nodeName,
		)
	}

	_ = response.WriteEntity(resp)
}