Job cm removal test & fix

pkg/components/config_helper.go

@@ -8,14 +8,16 @@ import (
 	"reflect"
 
 	"github.com/BurntSushi/toml"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/ytsaurus/yt-k8s-operator/pkg/apiproxy"
 	"github.com/ytsaurus/yt-k8s-operator/pkg/labeller"
 	"github.com/ytsaurus/yt-k8s-operator/pkg/resources"
 	"github.com/ytsaurus/yt-k8s-operator/pkg/ytconfig"
-	"go.ytsaurus.tech/yt/go/yson"
 	corev1 "k8s.io/api/core/v1"
+
+	"go.ytsaurus.tech/yt/go/yson"
 )
 
 const (
@@ -242,3 +244,11 @@ func (h *ConfigHelper) Fetch(ctx context.Context) error {
 	}
 	return h.configMap.Fetch(ctx)
 }
+
+func (h *ConfigHelper) RemoveConfigMapIfExists(ctx context.Context) error {
+	err := h.apiProxy.DeleteObject(ctx, h.configMap.OldObject())
+	if err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+	return nil
+}

pkg/components/init_job.go

@@ -181,6 +181,11 @@ func (j *InitJob) prepareRestart(ctx context.Context, dry bool) error {
 	if err := j.removeIfExists(ctx); err != nil {
 		return err
 	}
+
+	if err := j.configHelper.RemoveConfigMapIfExists(ctx); err != nil {
+		return err
+	}
+
 	j.conditionsManager.SetStatusCondition(metav1.Condition{
 		Type:    j.initCompletedCondition,
 		Status:  metav1.ConditionFalse,
@@ -191,7 +196,9 @@ func (j *InitJob) prepareRestart(ctx context.Context, dry bool) error {
 }
 
 func (j *InitJob) isRestartPrepared() bool {
-	return !resources.Exists(j.initJob) && j.conditionsManager.IsStatusConditionFalse(j.initCompletedCondition)
+	exists := resources.Exists(j.initJob)
+	isCondFalse := j.conditionsManager.IsStatusConditionFalse(j.initCompletedCondition)
+	return !exists && isCondFalse
 }
 
 func (j *InitJob) isRestartCompleted() bool {

pkg/components/init_job_test.go

@@ -0,0 +1,176 @@
+package components
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	ytv1 "github.com/ytsaurus/yt-k8s-operator/api/v1"
+	"github.com/ytsaurus/yt-k8s-operator/pkg/apiproxy"
+	"github.com/ytsaurus/yt-k8s-operator/pkg/consts"
+	"github.com/ytsaurus/yt-k8s-operator/pkg/labeller"
+	"github.com/ytsaurus/yt-k8s-operator/pkg/testutil"
+	"github.com/ytsaurus/yt-k8s-operator/pkg/ytconfig"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const (
+	ytsaurusName = "testsaurus"
+	domain       = "testdomain"
+
+	scriptBefore = "SCRIPT"
+	scriptAfter  = "UPDATED SCRIPT"
+)
+
+var (
+	waitTimeout = 5 * time.Second
+	waitTick    = 300 * time.Millisecond
+)
+
+func prepareTest(t *testing.T, namespace string) (*testutil.TestHelper, *apiproxy.Ytsaurus, *ytconfig.Generator) {
+	h := testutil.NewTestHelper(t, namespace, filepath.Join("..", "..", "config", "crd", "bases"))
+	h.Start(func(mgr ctrl.Manager) error { return nil })
+
+	ytsaurusResource := testutil.BuildMinimalYtsaurus(namespace, ytsaurusName)
+	// Deploy of ytsaurus spec is required, so it could set valid owner references for child resources.
+	testutil.DeployObject(h, &ytsaurusResource)
+
+	scheme := runtime.NewScheme()
+	utilruntime.Must(ytv1.AddToScheme(scheme))
+	fakeRecorder := record.NewFakeRecorder(100)
+
+	ytsaurus := apiproxy.NewYtsaurus(&ytsaurusResource, h.GetK8sClient(), fakeRecorder, scheme)
+	cfgen := ytconfig.NewGenerator(ytsaurus.GetResource(), domain)
+	return h, ytsaurus, cfgen
+}
+
+func syncJobUntilReady(t *testing.T, job *InitJob) {
+	ctx := context.Background()
+
+	require.Eventually(
+		t,
+		func() bool {
+			err := job.Fetch(ctx)
+			require.NoError(t, err)
+			st, err := job.Sync(ctx, false)
+			require.NoError(t, err)
+			return st.SyncStatus == SyncStatusReady
+		},
+		waitTimeout,
+		waitTick,
+	)
+}
+
+func newTestJob(ytsaurus *apiproxy.Ytsaurus) *InitJob {
+	k8sName := "dummy-name"
+	return NewInitJob(
+		&labeller.Labeller{
+			ObjectMeta: &metav1.ObjectMeta{
+				Name:      k8sName,
+				Namespace: ytsaurus.GetResource().Namespace,
+			},
+			ComponentLabel: "ms",
+			ComponentName:  k8sName,
+		},
+		ytsaurus.APIProxy(),
+		ytsaurus,
+		[]corev1.LocalObjectReference{},
+		"dummy",
+		consts.ClientConfigFileName,
+		"dummy-image",
+		func() ([]byte, error) { return []byte("dummy-cfg"), nil },
+	)
+}
+
+func TestJobRestart(t *testing.T) {
+	ctx := context.Background()
+
+	namespace := "testjobrestart"
+	h, ytsaurus, _ := prepareTest(t, namespace)
+	// TODO: separate helper so no need to remember to call stop
+	defer h.Stop()
+
+	job := newTestJob(ytsaurus)
+	job.SetInitScript(scriptBefore)
+	syncJobUntilReady(t, job)
+
+	err := job.prepareRestart(ctx, false)
+	require.NoError(t, err)
+
+	t.Log("Ensure job is deleted on restart")
+	require.Eventually(t,
+		func() bool {
+			batchJob := batchv1.Job{}
+			err = ytsaurus.APIProxy().Client().Get(ctx, client.ObjectKey{
+				Name:      "ms-init-job-dummy",
+				Namespace: namespace,
+			}, &batchJob)
+			return apierrors.IsNotFound(err)
+		},
+		waitTimeout,
+		100*time.Millisecond,
+	)
+
+	t.Log("Wait for restart to be prepared.")
+	require.Eventually(
+		t,
+		func() bool {
+			job = newTestJob(ytsaurus)
+			err = job.Fetch(ctx)
+			require.NoError(t, err)
+			return job.isRestartPrepared()
+		},
+		waitTimeout,
+		waitTick,
+	)
+}
+
+func TestJobScriptUpdateOnJobRestart(t *testing.T) {
+	ctx := context.Background()
+
+	namespace := "testjobscript"
+	h, ytsaurus, _ := prepareTest(t, namespace)
+	// TODO: separate helper so no need to remember to call stop
+	defer h.Stop()
+
+	job := newTestJob(ytsaurus)
+	job.SetInitScript(scriptBefore)
+	syncJobUntilReady(t, job)
+
+	err := job.prepareRestart(ctx, false)
+	require.NoError(t, err)
+
+	require.Eventually(
+		t,
+		func() bool {
+			job = newTestJob(ytsaurus)
+			err = job.Fetch(ctx)
+			require.NoError(t, err)
+			return job.isRestartPrepared()
+		},
+		waitTimeout,
+		waitTick,
+	)
+
+	// Imagine that new version of operator wants to set new init script for job.
+	job = newTestJob(ytsaurus)
+	job.SetInitScript(scriptAfter)
+	syncJobUntilReady(t, job)
+
+	cmData := testutil.FetchConfigMapData(
+		h,
+		"dummy-ms-init-job-config",
+		consts.InitClusterScriptFileName,
+	)
+	require.Equal(t, scriptAfter, cmData)
+}

pkg/components/suite_test.go

@@ -8,20 +8,21 @@ import (
 	"github.com/golang/mock/gomock"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"go.ytsaurus.tech/yt/go/yt"
 	appsv1 "k8s.io/api/apps/v1"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
+	"go.ytsaurus.tech/yt/go/yt"
+
 	mock_yt "github.com/ytsaurus/yt-k8s-operator/pkg/mock"
 )
 
-var ctrl *gomock.Controller
+var mockCtrl *gomock.Controller
 
 func TestComponents(t *testing.T) {
 	RegisterFailHandler(Fail)
 
-	ctrl = gomock.NewController(t)
+	mockCtrl = gomock.NewController(t)
 
 	RunSpecs(t, "Components fake suite")
 }

pkg/components/tablet_node_test.go

@@ -8,9 +8,6 @@ import (
 	"github.com/golang/mock/gomock"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"go.ytsaurus.tech/yt/go/guid"
-	"go.ytsaurus.tech/yt/go/ypath"
-	"go.ytsaurus.tech/yt/go/yt"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -20,6 +17,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
+	"go.ytsaurus.tech/yt/go/guid"
+	"go.ytsaurus.tech/yt/go/ypath"
+	"go.ytsaurus.tech/yt/go/yt"
+
 	ytv1 "github.com/ytsaurus/yt-k8s-operator/api/v1"
 	"github.com/ytsaurus/yt-k8s-operator/pkg/apiproxy"
 	mock_yt "github.com/ytsaurus/yt-k8s-operator/pkg/mock"
@@ -35,7 +36,7 @@ var _ = Describe("Tablet node test", func() {
 	var client client.WithWatch
 
 	BeforeEach(func() {
-		mockYtClient = mock_yt.NewMockClient(ctrl)
+		mockYtClient = mock_yt.NewMockClient(mockCtrl)
 
 		masterVolumeSize, _ := resource.ParseQuantity("1Gi")
 

pkg/testutil/testhelper.go

@@ -59,7 +59,7 @@ func NewTestHelper(t *testing.T, namespace, crdDirectoryPath string) *TestHelper
 		cancel:     testCancel,
 		k8sTestEnv: k8sTestEnv,
 		Namespace:  namespace,
-		ticker:     time.NewTicker(1 * time.Second),
+		ticker:     time.NewTicker(100 * time.Millisecond),
 	}
 }
 
@@ -182,6 +182,20 @@ func MarkAllJobsCompleted(h *TestHelper) {
 			job.Status.Succeeded = 1
 			UpdateObjectStatus(h, job)
 		}
+		if !job.DeletionTimestamp.IsZero() {
+			h.t.Logf(
+				"found job %s, with deletion ts %s and finalizers: %s. Deleting as gc would do in real cluster.",
+				job.Name,
+				job.DeletionTimestamp,
+				job.Finalizers,
+			)
+			job.Finalizers = []string{}
+			UpdateObject(h, &batchv1.Job{}, job)
+			err = h.k8sClient.Delete(context.Background(), job)
+			if err != nil && !apierrors.IsNotFound(err) {
+				panic(fmt.Sprintf("failed to delete job %s", job))
+			}
+		}
 	}
 }