Skip to content

Commit

Permalink
Adding validation checks for Bookkeeper Deployments (#598)
Browse files Browse the repository at this point in the history 
* Adding validation checks for Bookkeeper Deployments

Signed-off-by: Nishant Gupta <[email protected]>

* Adressing Review Comments

Signed-off-by: Nishant Gupta <[email protected]>
  • Loading branch information
@nishant-yt
nishant-yt authored Dec 7, 2021
1 parent e7af6d1 commit ff535f8
Show file tree
Hide file tree
Showing 4 changed files with 410 additions and 0 deletions.
15 changes: 15 additions & 0 deletions pkg/apis/pravega/v1beta1/pravega.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -383,6 +383,21 @@ func (s *PravegaSpec) withDefaults() (changed bool) {
s.Options = map[string]string{}
}

if s.Options["bookkeeper.ensemble.size"] == "" {
changed = true
s.Options["bookkeeper.ensemble.size"] = "3"
}

if s.Options["bookkeeper.write.quorum.size"] == "" {
changed = true
s.Options["bookkeeper.write.quorum.size"] = "3"
}

if s.Options["bookkeeper.ack.quorum.size"] == "" {
changed = true
s.Options["bookkeeper.ack.quorum.size"] = "3"
}

if s.ControllerJvmOptions == nil {
changed = true
s.ControllerJvmOptions = []string{}
Expand Down
74 changes: 74 additions & 0 deletions pkg/apis/pravega/v1beta1/pravegacluster_types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -905,6 +905,10 @@ func (p *PravegaCluster) ValidateCreate() error {
if err != nil {
return err
}
err = p.ValidateBookkeperSettings()
if err != nil {
return err
}
return nil
}

Expand All @@ -923,6 +927,10 @@ func (p *PravegaCluster) ValidateUpdate(old runtime.Object) error {
if err != nil {
return err
}
err = p.ValidateBookkeperSettings()
if err != nil {
return err
}
return nil
}

Expand Down Expand Up @@ -1296,6 +1304,72 @@ func (p *PravegaCluster) ValidateSegmentStoreMemorySettings() error {
return nil
}

// ValidateBookkeeperSettings checks that the value passed for the options bookkeeper.ensemble.size (E) bookkeeper.write.quorum.size (W)
// and bookkeeper.ack.quorum.size (A) adheres to the following rule, E >= W >= A.
// The method also checks for the option bookkeeper.write.quorum.racks.minimumCount.enable which should be set to false when bookkeeper.ensemble.size is 1.
// Note: The default value of E , W and A is 3.
func (p *PravegaCluster) ValidateBookkeperSettings() error {
// Intializing ensemble size, write quorum size and ack quorum size to default value of 3
ensembleSizeInt, writeQuorumSizeInt, ackQuorumSizeInt := 3, 3, 3
var err error

ensembleSize := p.Spec.Pravega.Options["bookkeeper.ensemble.size"]
writeQuorumSize := p.Spec.Pravega.Options["bookkeeper.write.quorum.size"]
ackQuorumSize := p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"]
writeQuorumRacks := p.Spec.Pravega.Options["bookkeeper.write.quorum.racks.minimumCount.enable"]

if len(ensembleSize) > 0 {
ensembleSizeInt, err = strconv.Atoi(ensembleSize)
if err != nil {
return fmt.Errorf("Cannot convert ensemble size from string to integer: %v", err)
}
}

if len(writeQuorumSize) > 0 {
writeQuorumSizeInt, err = strconv.Atoi(writeQuorumSize)
if err != nil {
return fmt.Errorf("Cannot convert write quorum size from string to integer: %v", err)
}
}

if len(ackQuorumSize) > 0 {
ackQuorumSizeInt, err = strconv.Atoi(ackQuorumSize)
if err != nil {
return fmt.Errorf("Cannot convert ack quorum size from string to integer: %v", err)
}
}

if writeQuorumRacks != "true" && writeQuorumRacks != "false" && writeQuorumRacks != "" {
return fmt.Errorf("bookkeeper.write.quorum.racks.minimumCount.enable can be only set to \"true\" \"false\" or \"\"")
}

if writeQuorumRacks == "true" && ensembleSizeInt == 1 {
return fmt.Errorf("bookkeeper.write.quorum.racks.minimumCount.enable should be set to false if bookkeeper.ensemble.size is 1")
}

if ensembleSizeInt < writeQuorumSizeInt {
if ensembleSize == "" {
return fmt.Errorf("The value provided for the option bookkeeper.write.quorum.size should be less than or equal to the value of option bookkeeper.ensemble.size (default is 3)")
}
if writeQuorumSize == "" {
return fmt.Errorf("The value provided for the option bookkeeper.ensemble.size should be greater than or equal to the value of option bookkeeper.write.quorum.size (default is 3)")
}
return fmt.Errorf("The value provided for the option bookkeeper.write.quorum.size should be less than or equal to the value of option bookkeeper.ensemble.size")
}

if writeQuorumSizeInt < ackQuorumSizeInt {
if writeQuorumSize == "" {
return fmt.Errorf("The value provided for the option bookkeeper.ack.quorum.size should be less than or equal to the value of option bookkeeper.write.quorum.size (default is 3)")
}
if ackQuorumSize == "" {
return fmt.Errorf("The value provided for the option bookkeeper.write.quorum.size should be greater than or equal to the value of option bookkeeper.ack.quorum.size (default is 3)")
}
return fmt.Errorf("The value provided for the option bookkeeper.ack.quorum.size should be less than or equal to the value of option bookkeeper.write.quorum.size")
}

return nil
}

//to return name of segmentstore based on the version
func (p *PravegaCluster) StatefulSetNameForSegmentstore() string {
if util.IsVersionBelow(p.Spec.Version, "0.7.0") {
Expand Down
212 changes: 212 additions & 0 deletions pkg/apis/pravega/v1beta1/pravegacluster_types_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -857,4 +857,216 @@ var _ = Describe("PravegaCluster Types Spec", func() {
})
})
})

Context("Validate Bookie Settings", func() {
var (
p *v1beta1.PravegaCluster
)

BeforeEach(func() {
p = &v1beta1.PravegaCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "default",
},
}
p.WithDefaults()
})

Context("Validating with correct values for Ensemble Size, Write Quorum size and Ack quorum size", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "3"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = "3"
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "3"
err = p.ValidateBookkeperSettings()
})

It("Should return nil", func() {
Ω(err).Should(BeNil())
})
})

Context("Invalid Value for Enseble size", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "3.4"
err = p.ValidateBookkeperSettings()
})

It("Should return error", func() {
Ω(strings.ContainsAny(err.Error(), "Cannot convert ensemble size from string to integer")).Should(Equal(true))
})
})

Context("Invalid Value for Write Quorum size", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = "3##4"
err = p.ValidateBookkeperSettings()
})

It("Should return error", func() {
Ω(strings.ContainsAny(err.Error(), "Cannot convert write quorum size from string to integer")).Should(Equal(true))
})
})

Context("Invalid Value for Ack Quorum size", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "2!342"
err = p.ValidateBookkeperSettings()
})

It("Should return error", func() {
Ω(strings.ContainsAny(err.Error(), "Cannot convert ack quorum size from string to integer")).Should(Equal(true))
})
})

Context("Validating with Ensemble Size < Write Quorum Size", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "3"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = "4"
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "3"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "The value provided for the option bookkeeper.write.quorum.size should be less than or equal to the value of option bookkeeper.ensemble.size")).Should(Equal(true))
})
})

Context("Validating with Ensemble size <=2 and Write quorum size is set to default", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "2"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = ""
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "3"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "The value provided for the option bookkeeper.ensemble.size should be greater than or equal to the value of option bookkeeper.write.quorum.size (default is 3)")).Should(Equal(true))
})
})

Context("Validating with Write quorum size > 3 and Ensemble size is set to default", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = ""
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = "4"
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "3"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "The value provided for the option bookkeeper.write.quorum.size should be less than or equal to the value of option bookkeeper.ensemble.size (default is 3)")).Should(Equal(true))
})
})

Context("Validating whether minimum racks count is set to true false or \"\"", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.write.quorum.racks.minimumCount.enable"] = "True"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "bookkeeper.write.quorum.racks.minimumCount.enable can be only set to \"true\" \"false\" or \"\"")).Should(Equal(true))
})
})

Context("Validating with Enseble Size set to 1 and minimum count enabled is set to true", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "1"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = ""
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = ""
p.Spec.Pravega.Options["bookkeeper.write.quorum.racks.minimumCount.enable"] = "true"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "bookkeeper.write.quorum.racks.minimumCount.enable should be set to false if bookkeeper.ensemble.size is 1")).Should(Equal(true))
})
})

Context("Validating with Write quorum size < Acq quorum size", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "4"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = "4"
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "5"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "The value provided for the option bookkeeper.ack.quorum.size should be less than or equal to the value of option bookkeeper.write.quorum.size")).Should(Equal(true))
})
})

Context("Validating with Write quorum size <=2 and Acq quorum size is set to default", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "3"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = "2"
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = ""
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "The value provided for the option bookkeeper.write.quorum.size should be greater than or equal to the value of option bookkeeper.ack.quorum.size (default is 3)")).Should(Equal(true))
})
})

Context("Validating with Ack quorum size > 3 and Write quorum size is set to default", func() {
var (
err error
)

BeforeEach(func() {
p.Spec.Pravega.Options["bookkeeper.ensemble.size"] = "3"
p.Spec.Pravega.Options["bookkeeper.write.quorum.size"] = ""
p.Spec.Pravega.Options["bookkeeper.ack.quorum.size"] = "4"
err = p.ValidateBookkeperSettings()
})

It("should return error", func() {
Ω(strings.ContainsAny(err.Error(), "The value provided for the option bookkeeper.ack.quorum.size should be less than or equal to the value of option bookkeeper.write.quorum.size (default is 3)")).Should(Equal(true))
})
})
})
})
Loading

0 comments on commit ff535f8

Please sign in to comment.