Merge pull request #5761 from vbnrh/rook-ac-controller-runtime
ceph: implement controller-runtime for admission controllers
This commit is contained in:
10 files changed
+332
-385
No files matched your search
@@ -18,7 +18,6 @@ package ceph
|
||||
|
||||
import (
|
||||
"github.com/rook/rook/cmd/rook/rook"
|
||||
"github.com/rook/rook/pkg/daemon/admission"
|
||||
operator "github.com/rook/rook/pkg/operator/ceph"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
@@ -36,11 +35,9 @@ func init() {
|
||||
|
||||
func startAdmissionController(cmd *cobra.Command, args []string) {
|
||||
rook.SetLogLevel()
|
||||
|
||||
rook.LogStartupInfo(admissionCmd.Flags())
|
||||
|
||||
context := rook.NewContext()
|
||||
|
||||
a := admission.New(context, "ceph", operator.ValidateCephResource)
|
||||
a.StartServer()
|
||||
err := operator.StartAdmissionController()
|
||||
if err != nil {
|
||||
rook.TerminateFatal(err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
/*
|
||||
Copyright 2020 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 v1
|
||||
|
||||
import (
|
||||
"github.com/pkg/errors"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/webhook"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// compile-time assertions ensures CephCluster implements webhook.Validator so a webhook builder
|
||||
// will be registered for the validating webhook.
|
||||
var _ webhook.Validator = &CephCluster{}
|
||||
|
||||
func (c *CephCluster) ValidateCreate() error {
|
||||
logger.Info("validate create cephcluster")
|
||||
//If external mode enabled, then check if other fields are empty
|
||||
if c.Spec.External.Enable {
|
||||
if c.Spec.Mon != (MonSpec{}) || c.Spec.Dashboard != (DashboardSpec{}) || c.Spec.Monitoring != (MonitoringSpec{}) || c.Spec.DisruptionManagement != (DisruptionManagementSpec{}) || len(c.Spec.Mgr.Modules) > 0 || len(c.Spec.Network.Provider) > 0 || len(c.Spec.Network.Selectors) > 0 {
|
||||
return errors.New("invalid create : external mode enabled cannot have mon,dashboard,monitoring,network,disruptionManagement,storage fields in CR")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *CephCluster) ValidateUpdate(old runtime.Object) error {
|
||||
logger.Info("validate update cephcluster")
|
||||
occ := old.(*CephCluster)
|
||||
return validateUpdatedCephCluster(c, occ)
|
||||
}
|
||||
|
||||
func (c *CephCluster) ValidateDelete() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateUpdatedCephCluster(updatedCephCluster *CephCluster, found *CephCluster) error {
|
||||
if updatedCephCluster.Spec.DataDirHostPath != found.Spec.DataDirHostPath {
|
||||
return errors.Errorf("invalid update: DataDirHostPath change from %q to %q is not allowed", found.Spec.DataDirHostPath, updatedCephCluster.Spec.DataDirHostPath)
|
||||
}
|
||||
|
||||
if updatedCephCluster.Spec.Network.HostNetwork != found.Spec.Network.HostNetwork {
|
||||
return errors.Errorf("invalid update: HostNetwork change from %q to %q is not allowed", strconv.FormatBool(found.Spec.Network.HostNetwork), strconv.FormatBool(updatedCephCluster.Spec.Network.HostNetwork))
|
||||
}
|
||||
|
||||
if updatedCephCluster.Spec.Network.Provider != found.Spec.Network.Provider {
|
||||
return errors.Errorf("invalid update: Provider change from %q to %q is not allowed", found.Spec.Network.Provider, updatedCephCluster.Spec.Network.Provider)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
/*
|
||||
Copyright 2020 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 v1
|
||||
|
||||
import (
|
||||
"github.com/coreos/pkg/capnslog"
|
||||
"github.com/pkg/errors"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/webhook"
|
||||
)
|
||||
|
||||
var (
|
||||
webhookName = "rook-ceph-webhook"
|
||||
logger = capnslog.NewPackageLogger("github.com/rook/rook", webhookName)
|
||||
)
|
||||
|
||||
var _ webhook.Validator = &CephBlockPool{}
|
||||
|
||||
func (p *CephBlockPool) ValidateCreate() error {
|
||||
logger.Infof("validate create cephblockpool %v", p)
|
||||
|
||||
err := ValidatePoolSpecs(p.Spec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func ValidatePoolSpecs(ps PoolSpec) error {
|
||||
// Checks if either ErasureCoded or Replicated fields are set
|
||||
if ps.ErasureCoded.CodingChunks <= 0 && ps.ErasureCoded.DataChunks <= 0 && ps.Replicated.TargetSizeRatio <= 0 && ps.Replicated.Size <= 0 {
|
||||
return errors.New("invalid create: either of erasurecoded or replicated fields should be set")
|
||||
}
|
||||
// Check if any of the ErasureCoded fields are populated. Then check if replicated is populated. Both can't be populated at same time.
|
||||
if ps.ErasureCoded.CodingChunks > 0 || ps.ErasureCoded.DataChunks > 0 || ps.ErasureCoded.Algorithm != "" {
|
||||
if ps.Replicated.Size > 0 || ps.Replicated.TargetSizeRatio > 0 {
|
||||
return errors.New("invalid create: both erasurecoded and replicated fields cannot be set at the same time")
|
||||
}
|
||||
}
|
||||
|
||||
if ps.Replicated.Size == 0 && ps.Replicated.TargetSizeRatio == 0 {
|
||||
// Check if datachunks is set and has value less than 2.
|
||||
if ps.ErasureCoded.DataChunks < 2 && ps.ErasureCoded.DataChunks != 0 {
|
||||
return errors.New("invalid create: erasurecoded.datachunks needs minimum value of 2")
|
||||
}
|
||||
|
||||
// Check if codingchunks is set and has value less than 1.
|
||||
if ps.ErasureCoded.CodingChunks < 1 && ps.ErasureCoded.CodingChunks != 0 {
|
||||
return errors.New("invalid create: erasurecoded.codingchunks needs minimum value of 1")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *CephBlockPool) ValidateUpdate(old runtime.Object) error {
|
||||
logger.Info("validate update cephblockpool")
|
||||
ocbp := old.(*CephBlockPool)
|
||||
err := ValidatePoolSpecs(p.Spec)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if p.Spec.ErasureCoded.CodingChunks > 0 || p.Spec.ErasureCoded.DataChunks > 0 || p.Spec.ErasureCoded.Algorithm != "" {
|
||||
if ocbp.Spec.Replicated.Size > 0 || ocbp.Spec.Replicated.TargetSizeRatio > 0 {
|
||||
return errors.New("invalid update: replicated field is set already in previous object. cannot be changed to use erasurecoded")
|
||||
}
|
||||
}
|
||||
|
||||
if p.Spec.Replicated.Size > 0 || p.Spec.Replicated.TargetSizeRatio > 0 {
|
||||
if ocbp.Spec.ErasureCoded.CodingChunks > 0 || ocbp.Spec.ErasureCoded.DataChunks > 0 || ocbp.Spec.ErasureCoded.Algorithm != "" {
|
||||
return errors.New("invalid update: erasurecoded field is set already in previous object. cannot be changed to use replicated")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *CephBlockPool) ValidateDelete() error {
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package v1
|
||||
|
||||
import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestCephClusterValidateCreate(t *testing.T) {
|
||||
c := &CephCluster{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "rook-ceph",
|
||||
},
|
||||
Spec: ClusterSpec{
|
||||
DataDirHostPath: "/var/lib/rook",
|
||||
},
|
||||
}
|
||||
err := c.ValidateCreate()
|
||||
assert.NoError(t, err)
|
||||
c.Spec.External.Enable = true
|
||||
c.Spec.Monitoring = MonitoringSpec{
|
||||
Enabled: true,
|
||||
RulesNamespace: "rook-ceph",
|
||||
}
|
||||
err = c.ValidateCreate()
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestValidatePoolSpec(t *testing.T) {
|
||||
p := &CephBlockPool{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "ec-pool",
|
||||
},
|
||||
Spec: PoolSpec{
|
||||
ErasureCoded: ErasureCodedSpec{
|
||||
CodingChunks: 1,
|
||||
DataChunks: 2,
|
||||
},
|
||||
},
|
||||
}
|
||||
err := ValidatePoolSpecs(p.Spec)
|
||||
assert.NoError(t, err)
|
||||
|
||||
p.Spec.ErasureCoded.DataChunks = 1
|
||||
err = ValidatePoolSpecs(p.Spec)
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestCephBlockPoolValidateUpdate(t *testing.T) {
|
||||
p := &CephBlockPool{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "ec-pool",
|
||||
},
|
||||
Spec: PoolSpec{
|
||||
Replicated: ReplicatedSpec{RequireSafeReplicaSize: true, Size: 3},
|
||||
},
|
||||
}
|
||||
up := p.DeepCopy()
|
||||
up.Spec.ErasureCoded.DataChunks = 2
|
||||
up.Spec.ErasureCoded.CodingChunks = 1
|
||||
err := up.ValidateUpdate(p)
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestCephClusterValidateUpdate(t *testing.T) {
|
||||
c := &CephCluster{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "rook-ceph",
|
||||
},
|
||||
Spec: ClusterSpec{
|
||||
DataDirHostPath: "/var/lib/rook",
|
||||
},
|
||||
}
|
||||
err := c.ValidateCreate()
|
||||
assert.NoError(t, err)
|
||||
|
||||
// Updating the CRD specs with invalid values
|
||||
uc := c.DeepCopy()
|
||||
uc.Spec.DataDirHostPath = "var/rook"
|
||||
err = uc.ValidateUpdate(c)
|
||||
assert.Error(t, err)
|
||||
}
|
||||
@@ -1,119 +0,0 @@
|
||||
/*
|
||||
Copyright 2020 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 admission
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/rook/rook/pkg/clusterd"
|
||||
"io/ioutil"
|
||||
"k8s.io/api/admission/v1beta1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
const (
|
||||
jsonContentType = `application/json`
|
||||
)
|
||||
|
||||
var (
|
||||
universalDeserializer = serializer.NewCodecFactory(runtime.NewScheme()).UniversalDeserializer()
|
||||
)
|
||||
|
||||
// admitFunc is a callback for admission controller logic. Given an AdmissionRequest, it returns an error that will be shown when the operation is rejected.
|
||||
type admitFunc func(*v1beta1.AdmissionRequest, *clusterd.Context) error
|
||||
|
||||
// doServeAdmitFunc parses the HTTP request for an admission controller webhook, and -- in case of a well-formed
|
||||
// request -- delegates the admission control logic to the given admitFunc. The response body is then returned as raw
|
||||
// bytes.
|
||||
func doServeAdmitFunc(w http.ResponseWriter, r *http.Request, a *AdmissionController) ([]byte, error) {
|
||||
// Step 1: Request validation. Only handle POST requests with a body and json content type.
|
||||
if r.Method != http.MethodPost {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return nil, errors.Errorf("invalid method %q, only POST requess are allowed", r.Method)
|
||||
}
|
||||
body, err := ioutil.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return nil, errors.Wrap(err, "could not read request body")
|
||||
}
|
||||
if contentType := r.Header.Get("Content-Type"); contentType != jsonContentType {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return nil, errors.Errorf("unsupported content type %q, only %q is supported", contentType, jsonContentType)
|
||||
}
|
||||
// Step 2: Parse the AdmissionReview request.
|
||||
var admissionReviewReq v1beta1.AdmissionReview
|
||||
if _, _, err := universalDeserializer.Decode(body, nil, &admissionReviewReq); err != nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return nil, errors.Wrap(err, "failed to deserialize request")
|
||||
} else if admissionReviewReq.Request == nil {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return nil, errors.New("malformed admission review. request is nil")
|
||||
}
|
||||
logger.Infof("processing webhook request for resource type %s", admissionReviewReq.Request.Kind.Kind)
|
||||
// Step 3: Construct the AdmissionReview response.
|
||||
admissionReviewResponse := v1beta1.AdmissionReview{
|
||||
Response: &v1beta1.AdmissionResponse{
|
||||
UID: admissionReviewReq.Request.UID,
|
||||
},
|
||||
}
|
||||
err = a.validator(admissionReviewReq.Request, a.context)
|
||||
if err != nil {
|
||||
// If the handler returned an error, incorporate the error message into the response and deny the object
|
||||
// creation.
|
||||
admissionReviewResponse.Response.Allowed = false
|
||||
admissionReviewResponse.Response.Result = &metav1.Status{
|
||||
Message: err.Error(),
|
||||
}
|
||||
} else {
|
||||
admissionReviewResponse.Response.Allowed = true
|
||||
}
|
||||
// Return the AdmissionReview with a response as JSON.
|
||||
bytes, err := json.Marshal(&admissionReviewResponse)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to marshal response to AdmissionReview")
|
||||
}
|
||||
return bytes, nil
|
||||
}
|
||||
|
||||
// serveAdmitFunc is a wrapper around doServeAdmitFunc that adds error handling and logging.
|
||||
func serveAdmitFunc(w http.ResponseWriter, r *http.Request, a *AdmissionController) {
|
||||
logger.Info("handling webhook request")
|
||||
|
||||
var writeErr error
|
||||
if bytes, err := doServeAdmitFunc(w, r, a); err != nil {
|
||||
logger.Errorf("failed to handle webhook request. %v", err)
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
_, writeErr = w.Write([]byte(err.Error()))
|
||||
} else {
|
||||
logger.Info("webhook request handled successfully")
|
||||
_, writeErr = w.Write(bytes)
|
||||
}
|
||||
|
||||
if writeErr != nil {
|
||||
logger.Errorf("failed to write response to webhook request. %v", writeErr)
|
||||
}
|
||||
}
|
||||
|
||||
// admitFuncHandler takes an admitFunc and wraps it into a http.Handler by means of calling serveAdmitFunc.
|
||||
func admitFuncHandler(a *AdmissionController) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
serveAdmitFunc(w, r, a)
|
||||
})
|
||||
}
|
||||
@@ -1,77 +0,0 @@
|
||||
/*
|
||||
Copyright 2020 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 admission
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"github.com/coreos/pkg/capnslog"
|
||||
"github.com/rook/rook/pkg/clusterd"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
const (
|
||||
tlsDir = "/etc/webhook"
|
||||
tlsCertFile = "tls.crt"
|
||||
tlsKeyFile = "tls.key"
|
||||
serverPort int32 = 8079
|
||||
)
|
||||
|
||||
var (
|
||||
logger = capnslog.NewPackageLogger("github.com/rook/rook", "admission")
|
||||
validatePath = "/validate"
|
||||
)
|
||||
|
||||
type AdmissionController struct {
|
||||
context *clusterd.Context
|
||||
providerName string
|
||||
validator admitFunc
|
||||
}
|
||||
|
||||
func New(context *clusterd.Context, providerName string, validator admitFunc) *AdmissionController {
|
||||
return &AdmissionController{
|
||||
context: context,
|
||||
providerName: providerName,
|
||||
validator: validator,
|
||||
}
|
||||
}
|
||||
|
||||
// StartServer will start the server
|
||||
func (a *AdmissionController) StartServer() {
|
||||
logger.Infof("starting the webhook for backend %q", a.providerName)
|
||||
certPath := filepath.Join(tlsDir, tlsCertFile)
|
||||
keyPath := filepath.Join(tlsDir, tlsKeyFile)
|
||||
keyPair, err := NewTlsKeypairReloader(certPath, keyPath)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to load certificate. %v", err)
|
||||
}
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle(validatePath, admitFuncHandler(a))
|
||||
|
||||
var httpServer *http.Server
|
||||
httpServer = &http.Server{
|
||||
Addr: fmt.Sprintf(":%d", serverPort),
|
||||
TLSConfig: &tls.Config{
|
||||
GetCertificate: keyPair.GetCertificateFunc(),
|
||||
},
|
||||
Handler: mux,
|
||||
}
|
||||
|
||||
httpServer.ListenAndServeTLS("", "")
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
/*
|
||||
Copyright (c) 2019 Multus 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.
|
||||
|
||||
Original code can be found at https://github.com/openshift/multus-admission-controller
|
||||
*/
|
||||
|
||||
package admission
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"github.com/pkg/errors"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
type tlsKeypairReloader struct {
|
||||
certMutex sync.RWMutex
|
||||
cert *tls.Certificate
|
||||
certPath string
|
||||
keyPath string
|
||||
}
|
||||
|
||||
func (keyPair *tlsKeypairReloader) maybeReload() error {
|
||||
newCert, err := tls.LoadX509KeyPair(keyPair.certPath, keyPair.keyPath)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to load the certificate key pair")
|
||||
}
|
||||
logger.Info("certificate reloaded")
|
||||
keyPair.certMutex.Lock()
|
||||
defer keyPair.certMutex.Unlock()
|
||||
keyPair.cert = &newCert
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetCertificateFunc will fetch the tls certificate
|
||||
func (keyPair *tlsKeypairReloader) GetCertificateFunc() func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
|
||||
return func(clientHello *tls.ClientHelloInfo) (*tls.Certificate, error) {
|
||||
keyPair.certMutex.RLock()
|
||||
defer keyPair.certMutex.RUnlock()
|
||||
return keyPair.cert, nil
|
||||
}
|
||||
}
|
||||
|
||||
// NewTlsKeyPairReloader will load the public and private key pair from given paths
|
||||
func NewTlsKeypairReloader(certPath, keyPath string) (*tlsKeypairReloader, error) {
|
||||
result := &tlsKeypairReloader{
|
||||
certPath: certPath,
|
||||
keyPath: keyPath,
|
||||
}
|
||||
cert, err := tls.LoadX509KeyPair(certPath, keyPath)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to load the certificate key pair")
|
||||
}
|
||||
result.cert = &cert
|
||||
|
||||
go func() {
|
||||
c := make(chan os.Signal, 1)
|
||||
signal.Notify(c, syscall.SIGHUP)
|
||||
for range c {
|
||||
if err := result.maybeReload(); err != nil {
|
||||
logger.Errorf("failed to reload certificate. %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
return result, nil
|
||||
}
|
||||
@@ -1,100 +0,0 @@
|
||||
/*
|
||||
Copyright 2020 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 operator
|
||||
|
||||
import (
|
||||
"github.com/pkg/errors"
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
|
||||
"github.com/rook/rook/pkg/clusterd"
|
||||
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
|
||||
"k8s.io/api/admission/v1beta1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
var (
|
||||
cephclusterResource = metav1.GroupVersionResource{Group: cephv1.CustomResourceGroup, Version: cephv1.Version, Resource: opcontroller.ClusterResource.Plural}
|
||||
rookClientSet rookclient.Interface
|
||||
universalDeserializer = serializer.NewCodecFactory(runtime.NewScheme()).UniversalDeserializer()
|
||||
)
|
||||
|
||||
func validateUpdatedCephCluster(updatedCephCluster cephv1.CephCluster, found *cephv1.CephCluster) error {
|
||||
if updatedCephCluster.Spec.DataDirHostPath != found.Spec.DataDirHostPath {
|
||||
return errors.Errorf("invalid update : DataDirHostPath change from %q to %q is not allowed", found.Spec.DataDirHostPath, updatedCephCluster.Spec.DataDirHostPath)
|
||||
}
|
||||
|
||||
if updatedCephCluster.Spec.Network.HostNetwork != found.Spec.Network.HostNetwork {
|
||||
return errors.Errorf("invalid update : HostNetwork change from %q to %q is not allowed", strconv.FormatBool(found.Spec.Network.HostNetwork), strconv.FormatBool(updatedCephCluster.Spec.Network.HostNetwork))
|
||||
}
|
||||
|
||||
if updatedCephCluster.Spec.Network.Provider != found.Spec.Network.Provider {
|
||||
return errors.Errorf("invalid update : Provider change from %q to %q is not allowed", found.Spec.Network.Provider, updatedCephCluster.Spec.Network.Provider)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func ValidateCephResource(request *v1beta1.AdmissionRequest, context *clusterd.Context) error {
|
||||
rookClientSet = context.RookClientset
|
||||
raw := request.Object.Raw
|
||||
reqCephCluster := cephv1.CephCluster{}
|
||||
// Deserializing the incoming CephCluster object
|
||||
if _, _, err := universalDeserializer.Decode(raw, nil, &reqCephCluster); err != nil {
|
||||
return errors.Wrap(err, "failed to deserialize pod object")
|
||||
}
|
||||
// Checks if CephCluster is being updated and rejects the request if the dataDirHostPath is changed from initial value
|
||||
if isCephClusterUpdate(request) {
|
||||
// Fetch the existing ceph cluster object
|
||||
found, err := fetchExistingCephCluster(request)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to fetch existing cephcluster")
|
||||
}
|
||||
err = validateUpdatedCephCluster(reqCephCluster, found)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to validate updated cephcluster")
|
||||
}
|
||||
}
|
||||
if isCephClusterCreate(request) {
|
||||
//If external mode enabled, then check if other fields are empty
|
||||
if reqCephCluster.Spec.External.Enable {
|
||||
if reqCephCluster.Spec.Mon != (cephv1.MonSpec{}) || reqCephCluster.Spec.Dashboard != (cephv1.DashboardSpec{}) || reqCephCluster.Spec.Monitoring != (cephv1.MonitoringSpec{}) || reqCephCluster.Spec.DisruptionManagement != (cephv1.DisruptionManagementSpec{}) || len(reqCephCluster.Spec.Mgr.Modules) > 0 || len(reqCephCluster.Spec.Network.Provider) > 0 || len(reqCephCluster.Spec.Network.Selectors) > 0 {
|
||||
return errors.New("invalid create : external mode enabled cannot have mon,dashboard,monitoring,network,disruptionManagement,storage fields in CR")
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isCephClusterCreate(request *v1beta1.AdmissionRequest) bool {
|
||||
return request.Resource == cephclusterResource && request.Operation == v1beta1.Create
|
||||
}
|
||||
|
||||
func fetchExistingCephCluster(request *v1beta1.AdmissionRequest) (*cephv1.CephCluster, error) {
|
||||
found, err := rookClientSet.CephV1().CephClusters(request.Namespace).Get(request.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return &cephv1.CephCluster{}, errors.Wrap(err, "failed to find existing cephcluster object")
|
||||
}
|
||||
return found, nil
|
||||
}
|
||||
|
||||
func isCephClusterUpdate(request *v1beta1.AdmissionRequest) bool {
|
||||
return request.Resource == cephclusterResource && request.Operation == v1beta1.Update
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
/*
|
||||
Copyright 2020 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 operator
|
||||
|
||||
import (
|
||||
"github.com/pkg/errors"
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
ctrl "sigs.k8s.io/controller-runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/webhook"
|
||||
)
|
||||
|
||||
type AdmissionController struct {
|
||||
providerName string
|
||||
}
|
||||
|
||||
var (
|
||||
scheme = runtime.NewScheme()
|
||||
resources = []webhook.Validator{&cephv1.CephCluster{}, &cephv1.CephBlockPool{}}
|
||||
)
|
||||
|
||||
const (
|
||||
// Default directory where TLS certs are stored
|
||||
certDir = "/etc/webhook"
|
||||
// Default port for server
|
||||
port = 8079
|
||||
)
|
||||
|
||||
// StartAdmissionController will start the server
|
||||
func StartAdmissionController() error {
|
||||
logger.Infof("starting the webhook for backend ceph")
|
||||
err := cephv1.AddToScheme(scheme)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to add to scheme")
|
||||
}
|
||||
opts := ctrl.Options{
|
||||
Scheme: scheme,
|
||||
Port: port,
|
||||
CertDir: certDir,
|
||||
}
|
||||
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), opts)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create manager")
|
||||
}
|
||||
for _, resource := range resources {
|
||||
err = ctrl.NewWebhookManagedBy(mgr).For(resource).Complete()
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to register webhooks")
|
||||
}
|
||||
}
|
||||
logger.Info("starting webhook server")
|
||||
err = mgr.Start(ctrl.SetupSignalHandler())
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to start server")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -9,12 +9,27 @@ webhooks:
|
||||
- apiGroups: ["ceph.rook.io"]
|
||||
apiVersions: ["v1"]
|
||||
operations: ["CREATE","UPDATE","DELETE"]
|
||||
resources: ["*"]
|
||||
resources: ["cephclusters"]
|
||||
clientConfig:
|
||||
service:
|
||||
name: ${SERVICE_NAME}
|
||||
namespace: ${NAMESPACE}
|
||||
path: /validate
|
||||
path: /validate-ceph-rook-io-v1-cephcluster
|
||||
caBundle: ${CA_BUNDLE}
|
||||
admissionReviewVersions: ["v1beta1"]
|
||||
sideEffects: None
|
||||
timeoutSeconds: 5
|
||||
- name: ${SERVICE_NAME}.${NAMESPACE}.svc
|
||||
rules:
|
||||
- apiGroups: ["ceph.rook.io"]
|
||||
apiVersions: ["v1"]
|
||||
operations: ["CREATE","UPDATE","DELETE"]
|
||||
resources: ["cephblockpools"]
|
||||
clientConfig:
|
||||
service:
|
||||
name: ${SERVICE_NAME}
|
||||
namespace: ${NAMESPACE}
|
||||
path: /validate-ceph-rook-io-v1-cephblockpool
|
||||
caBundle: ${CA_BUNDLE}
|
||||
admissionReviewVersions: ["v1beta1"]
|
||||
sideEffects: None
|
||||
|
||||
Reference in new issue
Block a user