[WIP] Fix the e2e tests of vineyard operator.

k8s/config/scheduler/config.yaml

@@ -9,6 +9,12 @@ leaderElection:
 profiles:
   - schedulerName: vineyard-scheduler
     plugins:
+      filter:
+        disabled:
+          - name: "*"
+      preScore:
+        disabled:
+          - name: "*"
       score:
         enabled:
           - name: Vineyard

k8s/test/e2e/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM ghcr.io/v6d-io/v6d/vineyard-python-dev:latest
+FROM ghcr.io/v6d-io/v6d/vineyard-python-dev:latest_x86_64
 
 WORKDIR /
 

k8s/test/e2e/assembly-demo/assembly-job1.py

@@ -31,7 +31,7 @@ def generate_df(index):
 stream = RecordBatchStream.new(vineyard_client)
 vineyard_client.persist(stream.id)
 print(stream.id)
-writer = stream.writer
+writer = stream.open_writer(vineyard_client)
 total_chunks = 10
 for idx in range(total_chunks):
     time.sleep(idx)

k8s/test/e2e/assembly-demo/assembly-local.py

@@ -24,7 +24,7 @@
 
 stream_id = env_dist['STREAM_ID']
 stream = vineyard_client.get(stream_id)
-reader = stream.reader
+reader = stream.open_reader(vineyard_client)
 
 index = 0
 global_meta = vineyard.ObjectMeta()

k8s/test/e2e/assembly-demo/distributed-job1.py

@@ -39,7 +39,7 @@ def generate_df(index):
 
 sys.stdout = sys.__stdout__
 print(meta.id, flush=True)
-writer = stream.writer
+writer = stream.open_writer(vineyard_client)
 total_chunks = 10
 for idx in range(total_chunks):
     time.sleep(idx)

k8s/test/e2e/assembly-demo/distributed-job2.py

@@ -46,7 +46,7 @@ def generate_df(index):
 tup = vineyard_client.create_metadata(meta)
 vineyard_client.persist(tup)
 
-writer = stream.writer
+writer = stream.open_writer(vineyard_client)
 total_chunks = 10
 for idx in range(total_chunks):
     time.sleep(idx)

k8s/test/e2e/deploy-raw-backup-and-recover/e2e.yaml

@@ -185,15 +185,15 @@ verify:
     - query: |
         kubectl get pod -l app=get-local-object -n vineyard-job -oname | \
           awk -F '/' '{print $2}' | \
-          head -n 1 | \
+          tail -n 1 | \
           xargs kubectl logs -n vineyard-job | \
           yq e '{"sum": .}' - | \
           yq e 'to_entries' -
       expected: ../verify/values.yaml
     - query: |
         kubectl get pod -l app=get-distributed-object -n vineyard-job -oname | \
           awk -F '/' '{print $2}' | \
-          head -n 1 | \
+          tail -n 1 | \
           xargs kubectl logs -n vineyard-job | \
           yq e '{"sum": .}' - | \
           yq e 'to_entries' -

k8s/test/e2e/serialize/e2e.yaml

@@ -50,6 +50,7 @@ verify:
           awk -F '/' '{print $2}' | \
           head -n 1 | \
           xargs kubectl logs -n vineyard-system | \
+          grep "test passed" | \
           yq e '{"result": .}' - | \
           yq e 'to_entries' -
       expected: ../verify/serialize.yaml

python/vineyard/io/tests/test_stream.py

@@ -47,7 +47,7 @@ def test_recordbatch_stream(vineyard_client):
     total_chunks = 10
 
     def producer(stream: RecordBatchStream, dtypes, produced: List):
-        writer = stream.writer
+        writer = stream.open_writer(vineyard_client)
         for idx in range(total_chunks):
             time.sleep(idx)
             chunk = generate_random_dataframe(dtypes, 2)  # np.random.randint(10, 100))
@@ -57,7 +57,7 @@ def producer(stream: RecordBatchStream, dtypes, produced: List):
         writer.finish()
 
     def consumer(stream: RecordBatchStream, produced: List):
-        reader = stream.reader
+        reader = stream.open_reader(vineyard_client)
         index = 0
         while True:
             try:

test/runner.py

@@ -1044,6 +1044,39 @@
         )
 
 
+def run_stream_test(meta, allocator, endpoints, test_args):
+    meta_prefix = 'vineyard_test_%s' % time.time()
+    metadata_settings = make_metadata_settings(meta, endpoints, meta_prefix)
+
+    with start_vineyardd(
+        metadata_settings,
+        ['--allocator', allocator],
+        default_ipc_socket=VINEYARD_CI_IPC_SOCKET,
+    ) as (_, rpc_socket_port):
+        start_time = time.time()
+        subprocess.check_call(
+            [
+                'pytest',
+                '-s',
+                '-vvv',
+                '--exitfirst',
+                '--durations=0',
+                '--log-cli-level',
+                'DEBUG',
+                'python/vineyard/io/tests',
+                *test_args,
+                '--vineyard-ipc-socket=%s' % VINEYARD_CI_IPC_SOCKET,
+                '--vineyard-endpoint=localhost:%s' % rpc_socket_port,
+            ],
+            cwd=os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'),
+        )
+        print(
+            'running python stream %s tests use %s seconds'
+            % (time.time() - start_time),
+            flush=True,
+        )
+
+
 def run_fuse_test(meta, allocator, endpoints, test_args):
     meta_prefix = 'vineyard_test_%s' % time.time()
     metadata_settings = make_metadata_settings(meta, endpoints, meta_prefix)
@@ -1054,23 +1087,23 @@
         default_ipc_socket=VINEYARD_CI_IPC_SOCKET,
     ) as (_, rpc_socket_port), start_fuse() as _:
         start_time = time.time()
         subprocess.check_call(
             [
                 'pytest',
                 '-s',
                 '-vvv',
                 '--exitfirst',
                 '--durations=0',
                 '--log-cli-level',
                 'DEBUG',
                 'modules/fuse/test',
                 *test_args,
                 '--vineyard-ipc-socket=%s' % VINEYARD_CI_IPC_SOCKET,
                 '--vineyard-endpoint=localhost:%s' % rpc_socket_port,
                 '--vineyard-fuse-mount-dir=%s' % VINEYARD_FUSE_MOUNT_DIR,
             ],
             cwd=os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'),
         )
         print(
             'running fuse tests use %s seconds' % (time.time() - start_time),
         )
@@ -1257,6 +1290,7 @@
 
     if args.with_io:
         run_io_adaptor_tests(args.meta, args.allocator, endpoints, python_test_args)
+        run_stream_test(args.meta, args.allocator, endpoints, python_test_args)
 
     if args.with_fuse:
         run_fuse_test(args.meta, args.allocator, endpoints, python_test_args)