Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(torch): processing duplications #3

Merged
merged 5 commits into from
Nov 13, 2023
Merged

fix(torch): processing duplications #3

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4398782
fix(torch): processing duplications
tty47 Nov 8, 2023
eddceca
fix(torch): refactor extract to func - check if pods are in Running s…
tty47 Nov 9, 2023
3e1661f
Merge https://github.com/celestiaorg/torch into jose/fix-processing-d…
tty47 Nov 10, 2023
aa62e48
fix(torch): check statements before continue
tty47 Nov 10, 2023
3c5a220
fix(torch): cleanup not needed comments
tty47 Nov 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions pkg/k8s/statefulsets.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@ import (
"github.com/celestiaorg/torch/pkg/db/redis"
)

const queueK8SNodes = "k8s"
const (
queueK8SNodes = "k8s" // queueK8SNodes name of the queue.
daNodePrefix = "da" // daNodePrefix name prefix that Torch will use to filter the StatefulSets.
)

// WatchStatefulSets watches for changes to the StatefulSets in the specified namespace and updates the metrics accordingly
func WatchStatefulSets() error {
Expand Down Expand Up @@ -43,10 +46,12 @@ func WatchStatefulSets() error {
// Watch for events on the watcher channel
for event := range watcher.ResultChan() {
if statefulSet, ok := event.Object.(*v1.StatefulSet); ok {
//log.Info("StatefulSet containers: ", statefulSet.Spec.Template.Spec.Containers)
if !ok {
log.Warn("Received an event that is not a StatefulSet. Skipping this resource...")
continue
}

// check if the node is DA, if so, send it to the queue to generate the multi address
if strings.HasPrefix(statefulSet.Name, "da") {
if isStatefulSetValid(statefulSet) {
err := redis.Producer(statefulSet.Name, queueK8SNodes)
if err != nil {
log.Error("ERROR adding the node to the queue: ", err)
Expand All @@ -58,3 +63,11 @@ func WatchStatefulSets() error {

return nil
}

// isStatefulSetValid validates the StatefulSet received.
// checks if the StatefulSet name contains the daNodePrefix, and if the StatefulSet is in the "Running" state.
func isStatefulSetValid(statefulSet *v1.StatefulSet) bool {
return strings.HasPrefix(statefulSet.Name, daNodePrefix) &&
statefulSet.Status.CurrentReplicas > 0 &&
statefulSet.Status.Replicas == statefulSet.Status.ReadyReplicas
}