Add schema map conflic pipeline

.github/workflows/tests.yaml

@@ -1218,3 +1218,24 @@ jobs:
       - name: Run reindexing chaos tests
         run: |
           ./debug-reindexing.sh
+  schema-map-conflict:
+    if: ${{ github.event.inputs.weaviate_max_major_version_test_set == 0 || github.event.inputs.weaviate_max_major_version_test_set >= 1.25 }}
+    name: Test that no race appears when deleting tenants while a node restars after schema updates.
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    env:
+      PERSISTENCE_LSM_ACCESS_STRATEGY: ${{inputs.lsm_access_strategy}}
+    steps:
+      - uses: actions/checkout@v3
+      - name: Login to Docker Hub
+        uses: docker/login-action@v2
+        with:
+          username: ${{secrets.DOCKER_USERNAME}}
+          password: ${{secrets.DOCKER_PASSWORD}}
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.22'
+      - name: Run schema conflict chaos tests
+        run: |
+          .schema-map-conflict.sh

apps/schema-map-conflict/Dockerfile

@@ -0,0 +1,17 @@
+FROM golang:1.21-alpine
+
+WORKDIR /workdir
+
+COPY go.mod go.sum ./
+
+RUN go mod download
+
+ARG APP_NAME
+ENV APP_NAME ${APP_NAME}
+
+COPY ./${APP_NAME}.go .
+COPY ./shared.go .
+
+RUN go build -o ${APP_NAME} .
+
+CMD "/workdir/${APP_NAME}"

apps/schema-map-conflict/cluster_healthy.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/weaviate/weaviate-go-client/v4/weaviate/data/replication"
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+	defer cancel()
+
+	log.Printf("Validate all objects were added, with consistency level ONE...")
+
+	objects := readObjectsFile("data/data.json")
+
+	for _, obj := range objects {
+		resp, err := randClient().Data().ObjectsGetter().
+			WithClassName(class.Class).WithTenant(obj.Tenant).
+			WithID(obj.ID.String()).
+			WithConsistencyLevel(replication.ConsistencyLevel.ONE).
+			Do(ctx)
+		if err != nil {
+			log.Fatalf("failed to query id %s: %v", obj.ID.String(), err)
+		}
+		if len(resp) == 0 || resp[0] == nil {
+			log.Fatalf("object %s not found", obj.ID.String())
+		}
+		if obj.ID != resp[0].ID {
+			log.Fatalf("expected object %s, got %s", obj.ID.String(), resp[0].ID)
+		}
+	}
+}

apps/schema-map-conflict/go.mod

@@ -0,0 +1,39 @@
+module replication
+
+go 1.21
+
+
+require (
+	github.com/go-openapi/strfmt v0.23.0
+	github.com/google/uuid v1.6.0
+	github.com/weaviate/weaviate v1.25.1
+	github.com/weaviate/weaviate-go-client/v4 v4.14.0
+)
+
+require (
+	github.com/PuerkitoBio/purell v1.1.1 // indirect
+	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
+	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
+	github.com/go-openapi/analysis v0.21.2 // indirect
+	github.com/go-openapi/errors v0.22.0 // indirect
+	github.com/go-openapi/jsonpointer v0.19.5 // indirect
+	github.com/go-openapi/jsonreference v0.19.6 // indirect
+	github.com/go-openapi/loads v0.21.1 // indirect
+	github.com/go-openapi/spec v0.20.4 // indirect
+	github.com/go-openapi/swag v0.22.3 // indirect
+	github.com/go-openapi/validate v0.21.0 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/oklog/ulid v1.3.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	go.mongodb.org/mongo-driver v1.14.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
+	golang.org/x/oauth2 v0.20.0 // indirect
+	golang.org/x/sys v0.19.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/grpc v1.64.0 // indirect
+	google.golang.org/protobuf v1.34.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

apps/schema-map-conflict/shared.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"math/rand"
+	"os"
+	"strconv"
+
+	client "github.com/weaviate/weaviate-go-client/v4/weaviate"
+	"github.com/weaviate/weaviate/entities/models"
+)
+
+var (
+	numTenants = 100
+	numClasses = 100
+
+	node1Client = client.New(client.Config{Scheme: "http", Host: "localhost:8080"})
+	node2Client = client.New(client.Config{Scheme: "http", Host: "localhost:8081"})
+	node3Client = client.New(client.Config{Scheme: "http", Host: "localhost:8082"})
+
+)
+
+func returnClass(index int) models.Class {
+	return models.Class{
+		Class: "MapConflict" + "_" + strconv.Itoa(index),
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled: true,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     "name",
+				DataType: []string{"string"},
+			},
+			{
+				Name:     "index",
+				DataType: []string{"int"},
+			},
+		},
+		ReplicationConfig: &models.ReplicationConfig{
+			Factor: 3,
+		},
+	}
+}
+
+func readObjectsFile(filename string) []*models.Object {
+	b, err := os.ReadFile(filename)
+	if err != nil {
+		log.Fatalf("failed to read objects file: %v", err)
+	}
+
+	var objects []*models.Object
+	err = json.Unmarshal(b, &objects)
+	if err != nil {
+		log.Fatalf("failed to unmarshal objects file: %v", err)
+	}
+
+	return objects
+}
+
+func randClient() *client.Client {
+	clients := []*client.Client{
+		node1Client,
+		node2Client,
+		node3Client,
+	}
+
+	return clients[rand.Intn(len(clients))]
+}
+
+func allClients() []*client.Client {
+	return []*client.Client{
+		node1Client,
+		node2Client,
+		node3Client,
+	}
+}
+
+func checkBatchInsertResult(created []models.ObjectsGetResponse, err error) {
+	if err != nil {
+		log.Fatalf("batch insert failed: %v", err)
+	}
+
+	for _, c := range created {
+		if c.Result != nil {
+			if c.Result.Errors != nil && c.Result.Errors.Error != nil {
+				log.Fatalf("failed to create obj: %+v, with status: %v",
+					c.Result.Errors.Error[0], c.Result.Status)
+			}
+		}
+	}
+}

apps/schema-map-conflict/tenant_creator.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"context"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/weaviate/weaviate/entities/models"
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+	defer cancel()
+
+	{
+		log.Println("Delete any existing data...")
+		err := node1Client.Schema().AllDeleter().Do(ctx)
+		if err != nil {
+			log.Fatalf("failed to delete all: %v", err)
+		}
+	}
+
+	{
+		log.Printf("Creating %s classes with %s tenants...", strconv.Itoa(numClasses), strconv.Itoa(numTenants))
+		for i := 0; i < numClasses; i++ {
+
+			class := returnClass(i)
+			err := node1Client.Schema().ClassCreator().WithClass(&class).Do(ctx)
+			if err != nil {
+				log.Fatalf("failed to create class %s: %v", class.Class, err)
+			}
+
+			for j := 0; j < numTenants; j++ {
+
+				log.Println("Creating tenants...")
+				err = node1Client.Schema().TenantsCreator().WithClassName(class.Class).WithTenants(models.Tenant{Name: "tenant" + strconv.Itoa(j)}).Do(ctx)
+				if err != nil {
+					log.Fatalf("failed to create tenants: %v", err)
+				}
+			}
+		}
+	}
+}

apps/schema-map-conflict/tenant_deleter.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"context"
+	"log"
+	"strconv"
+	"time"
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+	defer cancel()
+
+	{
+		log.Printf("Deleting %s classes with %s tenants...", strconv.Itoa(numClasses), strconv.Itoa(numTenants))
+		for i := 0; i < numClasses; i++ {
+
+			class := returnClass(i)
+			for j := 0; j < numTenants; j++ {
+
+				log.Println("Deleting tenants...")
+				err := node1Client.Schema().TenantsDeleter().WithClassName(class.Class).WithTenants("tenant" + strconv.Itoa(j)).Do(ctx)
+				if err != nil {
+					log.Fatalf("failed to delete tenants: %v", err)
+				}
+				// Sleep for a while to slow down the creation process
+				time.Sleep(1 * time.Second)
+			}
+
+			err := node1Client.Schema().ClassDeleter().WithClassName(class.Class).Do(ctx)
+			if err != nil {
+				log.Fatalf("failed to delete class %s: %v", class.Class, err)
+			}
+
+		}
+	}
+}

apps/schema-map-conflict/tenant_updater.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"context"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/weaviate/weaviate/entities/models"
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+	defer cancel()
+
+	{
+		log.Printf("Updating %s tenants from %s classes...", strconv.Itoa(numTenants), strconv.Itoa(numClasses))
+		for i := 0; i < numClasses; i++ {
+
+			class := returnClass(i)
+			tenants, err := node1Client.Schema().TenantsGetter().WithClassName(class.Class).Do(ctx)
+			if err != nil {
+				log.Fatalf("failed to get tenants: %v", err)
+			}
+
+			log.Println("Updating %s tenants...", strconv.Itoa(len(tenants)))
+			for _, tenant := range tenants {
+				var desiredStatus string
+				if tenant.ActivityStatus == "HOT" {
+					desiredStatus = "COLD"
+				} else {
+					desiredStatus = "HOT"
+				}
+				err := node1Client.Schema().TenantsUpdater().WithClassName(class.Class).WithTenants(models.Tenant{Name: tenant.Name, ActivityStatus: desiredStatus}).Do(ctx)
+				if err != nil {
+					log.Fatalf("failed to update tenant %v: %v", tenant.Name, err)
+				}
+				// Sleep for a while to slow down the update process
+				time.Sleep(1 * time.Second)
+			}
+		}
+	}
+}

schema-map-conflict.sh

@@ -0,0 +1,43 @@
+#!/bin/bash
+
+set -e
+
+source common.sh
+
+echo "Building all required containers"
+( cd apps/schema-map-conflict/ && docker build --build-arg APP_NAME=tenant_creator -t tenant_creator . )
+( cd apps/schema-map-conflict/ && docker build --build-arg APP_NAME=tenant_updater -t tenant_updater . )
+( cd apps/schema-map-conflict/ && docker build --build-arg APP_NAME=tenant_deleter -t tenant_deleter . )
+
+rm -rf workdir
+mkdir workdir
+touch workdir/data.json
+
+echo "Starting Weaviate..."
+docker compose -f apps/weaviate/docker-compose-replication_single_voter.yml up -d weaviate-node-1 weaviate-node-2 weaviate-node-3
+wait_weaviate 8080
+wait_weaviate 8081
+wait_weaviate 8082
+
+echo "Killing node 3"
+docker compose -f apps/weaviate/docker-compose-replication_single_voter.yml kill weaviate-node-3
+
+
+# Create multi tenant classes while the node is down
+docker run  --network host -v "$PWD/workdir/data.json:/workdir/data.json" --name tenant_creator -t tenant_creator
+
+# Start updating tenants in the background
+# Create multi tenant classes while the node is down
+docker run -d --network host -v "$PWD/workdir/data.json:/workdir/data.json" --name tenant_updater -t tenant_updater
+
+# Create multi tenant classes while the node is down
+docker run -d --network host -v "$PWD/workdir/data.json:/workdir/data.json" --name tenant_deleter -t tenant_deleter 
+
+# Restart dead node, read objects from Node 3 while tenants are being deleted
+echo "Restart node 3"
+docker compose -f apps/weaviate/docker-compose-replication_single_voter.yml up -d weaviate-node-3
+wait_weaviate 8082
+
+
+echo "Success!"
+shutdown