add 0.9.4 to the tests and use tags

.github/workflows/flink.yaml

@@ -23,11 +23,35 @@ jobs:
       github.event.label.name == 'test-flink' ||
       contains( github.event.pull_request.labels.*.name, 'test-all') ||
       contains( github.event.pull_request.labels.*.name, 'test-flink')
-
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: [ "3.9", "3.10", "3.11" ]
+        recipes-version: [
+            "pangeo-forge-recipes==0.9.4",
+            "pangeo-forge-recipes==0.10.0",
+            "pangeo-forge-recipes==0.10.3",
+        ]
+        beam-version: [
+            "apache-beam==2.47.0",
+            # save some cycles and infer it might
+            # work on all versions between lowest and highest
+            # "apache-beam==2.48.0",
+            # "apache-beam==2.49.0",
+            # "apache-beam==2.50.0",
+            "apache-beam==2.51.0",
+        ]
+        # keep here to be explicit for future users what version
+        # of Flink we are supporting
+        flink-version: [ "1.16", ]
 
     steps:
     - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
 
     # Starts a k8s cluster with NetworkPolicy enforcement and installs both
     # kubectl and helm
@@ -64,43 +88,45 @@ jobs:
     - name: 'Set up Cloud SDK'
       uses: 'google-github-actions/setup-gcloud@v0'
 
-
-    - name: Set up Python 3.9
-      uses: actions/setup-python@v3
-      with:
-        python-version: 3.9
-
-    - name: 'Setup minio + mc'
+    - name: 'Setup minio cli mc'
       run: |
-        wget --quiet https://dl.min.io/server/minio/release/linux-amd64/minio
-        chmod +x minio
-        mv minio /usr/local/bin/minio
-
         wget --quiet https://dl.min.io/client/mc/release/linux-amd64/mc
         chmod +x mc
         mv mc /usr/local/bin/mc
-
-        minio --version
         mc --version
 
+    - name: Setup apache-beam "${{ matrix.beam-version }}" to satisfy pf-runner installs
+      run: |
+        pip install ${{ matrix.beam-version }}
 
     - name: Install dependencies & our package
       run: |
         python -m pip install --upgrade pip
         python -m pip install -r dev-requirements.txt
-        python -m pip install -e .
+        python -m pip install -e .[flink]
 
+    - name: Set up min.io as a k3s service
+      run: |
+        MYACCESSKEY=$(openssl rand -hex 16)
+        MYSECRETKEY=$(openssl rand -hex 16)
+        kubectl create secret generic minio-secrets --from-literal=MINIO_ACCESS_KEY=$MYACCESSKEY --from-literal=MINIO_SECRET_KEY=$MYSECRETKEY
+        kubectl apply -f tests/integration/flink/minio-manifest.yaml
 
     - name: Install socat so kubectl port-forward will work
       run: |
         # Not sure if this is why kubectl proxy isn't working, but let's try
         sudo apt update --yes && sudo apt install --yes socat
 
+#    - name: Setup upterm session
+#      uses: lhotari/action-upterm@v1
+
     - name: Test with pytest
       run: |
-        pytest -vvv -s --cov=pangeo_forge_runner tests/integration/test_flink.py
-        kubectl get pod -A
-        kubectl describe pod
+        beamversion=$(echo ${{ matrix.beam-version }} | cut -d"=" -f3)
+        pytest -vvv -s --cov=pangeo_forge_runner tests/integration/flink/test_flink_integration.py \
+          --flinkversion=${{ matrix.flink-version }} \
+          --pythonversion=${{ matrix.python-version }} \
+          --beamversion=$beamversion
 
     - name: Upload Coverage to Codecov
       uses: codecov/codecov-action@v2

setup.py

@@ -18,9 +18,12 @@
         "escapism",
         "jsonschema",
         "traitlets",
-        "apache-beam[gcp]",
         "importlib-metadata",
     ],
+    extras_require={
+        "dataflow": ["apache-beam[gcp]"],
+        "flink": ["apache-beam>=2.47.0"],
+    },
     entry_points={
         "console_scripts": ["pangeo-forge-runner=pangeo_forge_runner.cli:main"]
     },

tests/conftest.py

@@ -1,3 +1,4 @@
+import base64
 import os
 import secrets
 import signal
@@ -57,3 +58,66 @@ def minio(local_ip):
         proc.wait()
 
         assert proc.returncode == 0
+
+
+@pytest.fixture(scope="session")
+def minio_service():
+    cmd = [
+        "kubectl",
+        "get",
+        "service/minio-service",
+        "-o=jsonpath='{.spec.clusterIP}:{.spec.ports[0].port}'",
+    ]
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    assert proc.returncode == 0
+    svc_address = proc.stdout.strip('"').strip("'")
+    endpoint = f"http://{svc_address}"
+
+    cmd = [
+        "kubectl",
+        "get",
+        "secret/minio-secrets",
+        "-o=jsonpath='{.data.MINIO_ACCESS_KEY}'",
+    ]
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    assert proc.returncode == 0
+    myaccesskey = proc.stdout
+    myaccesskey = base64.b64decode(myaccesskey).decode()
+
+    cmd = [
+        "kubectl",
+        "get",
+        "secret/minio-secrets",
+        "-o=jsonpath='{.data.MINIO_SECRET_KEY}'",
+    ]
+    proc = subprocess.run(cmd, capture_output=True, text=True)
+    assert proc.returncode == 0
+    mysecretkey = proc.stdout
+    mysecretkey = base64.b64decode(mysecretkey).decode()
+
+    # enter
+    yield {"endpoint": endpoint, "username": myaccesskey, "password": mysecretkey}
+
+    # exit
+    return
+
+
+def pytest_addoption(parser):
+    parser.addoption("--flinkversion", action="store", default="1.16")
+    parser.addoption("--pythonversion", action="store", default="3.9")
+    parser.addoption("--beamversion", action="store", default="2.47.0")
+
+
+@pytest.fixture
+def flinkversion(request):
+    return request.config.getoption("--flinkversion")
+
+
+@pytest.fixture
+def pythonversion(request):
+    return request.config.getoption("--pythonversion")
+
+
+@pytest.fixture
+def beamversion(request):
+    return request.config.getoption("--beamversion")

tests/integration/flink/minio-manifest.yaml

@@ -0,0 +1,68 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: minio-pv-claim
+  labels:
+    app: minio-storage-claim
+spec:
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: 1Gi
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: minio
+spec:
+  selector:
+    matchLabels:
+      app: minio-server
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app: minio-server
+    spec:
+      containers:
+      - name: minio
+        image: minio/minio
+        args:
+        - server
+        - /data
+        env:
+        - name: MINIO_ACCESS_KEY
+          valueFrom:
+            secretKeyRef:
+              name: minio-secrets
+              key: MINIO_ACCESS_KEY
+        - name: MINIO_SECRET_KEY
+          valueFrom:
+            secretKeyRef:
+              name: minio-secrets
+              key: MINIO_SECRET_KEY
+        ports:
+        - containerPort: 9000
+          name: minio
+        volumeMounts:
+        - name: data
+          mountPath: "/data"
+      volumes:
+        - name: data
+          persistentVolumeClaim:
+            claimName: minio-pv-claim
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: minio-service
+spec:
+  type: ClusterIP
+  ports:
+    - port: 9000
+      targetPort: 9000
+      protocol: TCP
+  selector:
+    app: minio-server