Drop all tables if current schema version is not equal 0.2.0

cmd/icinga-kubernetes/main.go

@@ -5,9 +5,11 @@ import (
 	"flag"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/icinga/icinga-kubernetes/internal"
+	"github.com/icinga/icinga-kubernetes/pkg/backoff"
 	"github.com/icinga/icinga-kubernetes/pkg/com"
 	"github.com/icinga/icinga-kubernetes/pkg/database"
 	"github.com/icinga/icinga-kubernetes/pkg/periodic"
+	"github.com/icinga/icinga-kubernetes/pkg/retry"
 	schemav1 "github.com/icinga/icinga-kubernetes/pkg/schema/v1"
 	"github.com/icinga/icinga-kubernetes/pkg/sync"
 	syncv1 "github.com/icinga/icinga-kubernetes/pkg/sync/v1"
@@ -23,6 +25,8 @@ import (
 	"time"
 )
 
+const expectedSchemaVersion = "0.2.0"
+
 func main() {
 	runtime.ReallyCrash = true
 
@@ -71,6 +75,50 @@ func main() {
 		klog.Fatal(err)
 	}
 
+	g, ctx := errgroup.WithContext(context.Background())
+
+	var version string
+	err = retry.WithBackoff(
+		ctx,
+		func(ctx context.Context) (err error) {
+			query := "SELECT version FROM kubernetes_schema ORDER BY id DESC LIMIT 1"
+			err = db.QueryRowxContext(ctx, query).Scan(&version)
+			if err != nil {
+				err = database.CantPerformQuery(err, query)
+			}
+			return
+		},
+		retry.Retryable,
+		backoff.NewExponentialWithJitter(128*time.Millisecond, 1*time.Minute),
+		retry.Settings{})
+	if err != nil {
+		klog.Fatal(err)
+	}
+
+	if hasSchema && version != expectedSchemaVersion {
+		rows, err := db.Query(
+			db.Rebind("SELECT table_name FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=?"),
+			d.Database,
+		)
+		if err != nil {
+			klog.Fatal(err)
+		}
+		defer rows.Close()
+
+		dbLog.Info("Dropping schema")
+		for rows.Next() {
+			var tableName string
+			if err := rows.Scan(&tableName); err != nil {
+				klog.Fatal(err)
+			}
+			_, err := db.Exec("DROP TABLE " + tableName)
+			if err != nil {
+				klog.Fatal(err)
+			}
+		}
+		hasSchema = false
+	}
+
 	if !hasSchema {
 		dbLog.Info("Importing schema")
 
@@ -83,7 +131,6 @@ func main() {
 		}
 	}
 
-	g, ctx := errgroup.WithContext(context.Background())
 	g.Go(func() error {
 		s := syncv1.NewSync(db, factory.Core().V1().Namespaces().Informer(), log.WithName("namespaces"), schemav1.NewNamespace)