Fix the e2e test in vineyard operator.

Signed-off-by: Ye Cao <[email protected]>

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_io_adaptor_tests(meta, allocator, endpoints, test_args):
  )
 
 
+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)
@@ -1257,6 +1290,7 @@ def execute_tests(args):
 
  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)