Invoke plugins lambda from backend for npe2 manifest discovery

.happy/terraform/modules/ecs-stack/main.tf

@@ -117,6 +117,7 @@ module backend_lambda {
     "DD_ENV" = var.env
     "DD_SERVICE" = local.custom_stack_name
     "API_URL" = var.env == "dev" ? module.api_gateway_proxy_stage.invoke_url : ""
+    "PLUGINS_LAMBDA_NAME" = local.plugins_function_name
   }
 
   log_retention_in_days = 14
@@ -144,8 +145,8 @@ module plugins_lambda {
 
   log_retention_in_days = 14
   timeout               = 150
-  memory_size           = 10240
-  ephemeral_storage_size = 10240
+  memory_size           = 256
+  ephemeral_storage_size = 512
 
 }
 
@@ -213,19 +214,36 @@ data aws_iam_policy_document backend_policy {
 
     resources = ["${local.data_bucket_arn}/*"]
   }
+
+  statement {
+    actions = [
+      "lambda:InvokeFunction"
+    ]
+
+    resources = [
+      module.plugins_lambda.function_arn,
+    ]
+  }
 }
 
 data aws_iam_policy_document plugins_policy {
   statement {
     actions = [
       "s3:PutObject",
       "s3:GetObject",
-      "s3:DeleteObject",
     ]
 
     resources = ["${local.data_bucket_arn}/*"]
   }
 
+  statement {
+    actions = [
+      "s3:ListBucket",
+    ]
+
+    resources = ["${local.data_bucket_arn}"]
+  }
+
 }
 
 resource aws_iam_role_policy policy {

backend/api/app.py

@@ -69,24 +69,23 @@ def versioned_plugin(plugin: str, version: str = None) -> Response:
 @app.route('/manifest/<plugin>', defaults={'version': None})
 @app.route('/manifest/<plugin>/versions/<version>')
 def plugin_manifest(plugin: str, version: str = None) -> Response:
-    max_failure_tries = 2
     manifest = get_manifest(plugin, version)
 
     if not manifest:
         return app.make_response(("Plugin does not exist", 404))
 
-    if 'process_count' not in manifest:
+    if 'error' not in manifest:
         return jsonify(manifest)
 
-    current_tries = manifest['process_count']
-    if current_tries >= max_failure_tries:
-        return app.make_response(
-            ("Plugin Manifest Not Found. Installation failed or plugin does not implement npe2", 404))
-    else:
+    error = manifest['error']
+    if error == 'Manifest not yet processed.':
         response = app.make_response(("Temporarily Unavailable. Attempting to build manifest. Please check back"
                                       " in 5 minutes.", 503))
         response.headers["Retry-After"] = 300
         return response
+    else:
+        return app.make_response(
+            ("Plugin Manifest Not Found. Manifest discovery failed.", 404))
 
 
 @app.route('/shields/<plugin>')

backend/api/model.py

@@ -1,5 +1,7 @@
 from concurrent import futures
 from datetime import datetime
+import json
+import os
 from typing import Tuple, Dict, List, Callable, Any
 from zipfile import ZipFile
 from io import BytesIO
@@ -12,6 +14,7 @@
 from utils.utils import render_description, send_alert, get_attribute, get_category_mapping, parse_manifest
 from utils.datadog import report_metrics
 from api.zulip import notify_new_packages
+import boto3
 
 index_subset = {'name', 'summary', 'description_text', 'description_content_type',
                 'authors', 'license', 'python_version', 'operating_system',
@@ -74,13 +77,36 @@ def get_plugin(plugin: str, version: str = None) -> dict:
 
 
 def get_frontend_manifest_metadata(plugin, version):
-    # load manifest from json (triggering build)
+    """Get manifest from cache, if it exists and parse into frontend fields
+
+    When `error` is in the returned metadata, we return
+    default values to the frontend.
+
+    :param plugin: name of the plugin to get
+    :param version: version of the plugin manifest
+    :return: parsed metadata for the frontend
+    """
     raw_metadata = get_manifest(plugin, version)
-    if 'process_count' in raw_metadata:
+    if 'error' in raw_metadata:
         raw_metadata = None
     interpreted_metadata = parse_manifest(raw_metadata)
     return interpreted_metadata
 
+def discover_manifest(plugin: str, version: str = None):
+    """
+    Invoke plugins lambda to generate manifest & write to cache.
+
+    :param plugin: name of the plugin to get
+    :param version: version of the plugin manifest
+    """
+    client = boto3.client('lambda')
+    lambda_event = {'plugin': plugin, 'version': version}
+    # this lambda invocation will call `napari-hub/plugins/get_plugin_manifest/generate_manifest`
+    client.invoke(
+        FunctionName=os.environ.get('PLUGINS_LAMBDA_NAME'),
+        InvocationType='Event',
+        Payload=json.dumps(lambda_event),
+    )
 
 def get_manifest(plugin: str, version: str = None) -> dict:
     """
@@ -95,11 +121,21 @@ def get_manifest(plugin: str, version: str = None) -> dict:
     elif version is None:
         version = plugins[plugin]
     plugin_metadata = get_cache(f'cache/{plugin}/{version}-manifest.json')
-    if plugin_metadata:
-        return plugin_metadata
-    else:
-        cache({"process_count": 0}, f'cache/{plugin}/{version}-manifest.json')
-        return {"process_count": 0}
+
+    # plugin_metadata being None indicates manifest is not cached and needs processing 
+    if plugin_metadata is None:
+        return {'error': 'Manifest not yet processed.'}
+
+    # empty dict indicates some lambda error in processing e.g. timed out
+    if plugin_metadata == {}:
+        return {'error': 'Processing manifest failed due to external error.'}
+
+    # error written to file indicates manifest discovery failed
+    if 'error' in plugin_metadata:
+        return {'error': plugin_metadata['error']}
+
+    # correct plugin manifest
+    return plugin_metadata
 
 
 def get_index() -> dict:
@@ -137,7 +173,15 @@ def get_excluded_plugins() -> Dict[str, str]:
 
 
 def build_manifest_metadata(plugin: str, version: str) -> Tuple[str, dict]:
-    metadata = get_frontend_manifest_metadata(plugin, version)
+    manifest = get_manifest(plugin, version)
+    if 'error' in manifest:
+        if 'Manifest not yet processed' in manifest['error']:
+            # this will invoke the plugins lambda & write manifes to cache
+            discover_manifest(plugin, version)
+        # return just default values for now
+        metadata = parse_manifest()
+    else:
+        metadata = parse_manifest(manifest)
     return plugin, metadata
 
 

frontend/src/utils/featureFlags.ts

@@ -24,7 +24,7 @@ export const FEATURE_FLAGS = createFeatureFlags({
   },
 
   npe2: {
-    environments: ['dev'],
+    environments: ['dev', 'staging'],
   },
 
   collections: {

plugins/_tests/test_discovery.py

@@ -2,20 +2,18 @@
 import os
 import pytest
 from unittest import mock
+from unittest.mock import call
 from plugins.get_plugin_manifest import generate_manifest
 from botocore.exceptions import ClientError
 
 TEST_PLUGIN = 'test-plugin'
 TEST_VERSION = '0.0.1'
 TEST_BUCKET = 'test-bucket'
-TEST_CACHE_PATH = f'cache/{TEST_PLUGIN}/{TEST_PLUGIN}.{TEST_VERSION}-manifest.json'
+TEST_CACHE_PATH = f'cache/{TEST_PLUGIN}/{TEST_VERSION}-manifest.json'
 
-def _mock_put_object(Body, Bucket, Key):
-    if os.path.exists(Key):
-        raise FileExistsError
-    else:
-        with open(Key, 'w') as fp:
-            fp.write(Body)
+def _mock_put_object(Body, Key):
+    with open(Key, 'w') as fp:
+        fp.write(Body)
 
 
 @mock.patch('plugins.get_plugin_manifest.s3')
@@ -30,7 +28,7 @@ def test_discovery_manifest_exists(s3, tmp_path):
         bucket_instance = s3.Bucket.return_value
         bucket_instance.objects.filter.return_value = [TEST_PLUGIN]
         generate_manifest({'plugin': TEST_PLUGIN, 'version': TEST_VERSION}, None)
-    s3.put_object.assert_not_called()
+    bucket_instance.put_object.assert_not_called()
 
 
 @mock.patch('plugins.get_plugin_manifest.s3')
@@ -67,7 +65,7 @@ def test_discovery_failure(s3, tmp_path):
     with mock.patch('plugins.get_plugin_manifest.bucket_path', tmp_path):
         bucket_instance = s3.Bucket.return_value
         bucket_instance.objects.filter.return_value = []
-        s3.put_object = _mock_put_object
+        bucket_instance.put_object = _mock_put_object
         generate_manifest({'plugin': TEST_PLUGIN, 'version': TEST_VERSION}, None)
     written = json.loads(manifest_pth.read_text())
     assert written['error'] == 'HTTP Error 404: Not Found'
@@ -80,17 +78,33 @@ def test_discovery_success(s3, tmp_path):
     plugin_name = 'napari-demo'
     plugin_version = 'v0.1.0'
 
-    manifest_pth = tmp_path / f'cache/{plugin_name}/{plugin_name}.{plugin_version}-manifest.json'
+    manifest_pth = tmp_path / f'cache/{plugin_name}/{plugin_version}-manifest.json'
     manifest_pth.parent.mkdir(parents=True)
     with mock.patch('plugins.get_plugin_manifest.bucket_path', tmp_path):
         bucket_instance = s3.Bucket.return_value
         bucket_instance.objects.filter.return_value = []
-        s3.put_object = _mock_put_object
+        bucket_instance.put_object = _mock_put_object
         generate_manifest({'plugin': plugin_name, 'version': plugin_version}, None)
     written = json.loads(manifest_pth.read_text())
     assert written['name'] == 'napari-demo'    
     assert len(written['contributions']['widgets']) == 1
 
+@mock.patch('plugins.get_plugin_manifest.bucket_name', '')
 def test_bucket_name_not_set():
     with pytest.raises(RuntimeError, match='Bucket name not specified.'):
         generate_manifest({}, None)
+
+@mock.patch('plugins.get_plugin_manifest.s3')
+@mock.patch('plugins.get_plugin_manifest.bucket_name', 'napari-hub')
+def test_file_always_written(s3, tmp_path):
+    plugin_name = 'napari-demo'
+    plugin_version = 'v0.1.0'
+
+    manifest_pth = tmp_path / f'cache/{plugin_name}/{plugin_version}-manifest.json'
+    manifest_pth.parent.mkdir(parents=True)
+    with mock.patch('plugins.get_plugin_manifest.bucket_path', tmp_path):
+        bucket_instance = s3.Bucket.return_value
+        bucket_instance.objects.filter.return_value = []
+        generate_manifest({'plugin': plugin_name, 'version': plugin_version}, None)
+    assert bucket_instance.put_object.call_count == 2
+    bucket_instance.put_object.assert_has_calls([call(Body='{}', Key=str(manifest_pth))])

plugins/get_plugin_manifest.py

@@ -18,16 +18,24 @@ def generate_manifest(event, context):
 
     plugin = event['plugin']
     version = event['version']
-    key = os.path.join(bucket_path, f'cache/{plugin}/{plugin}.{version}-manifest.json')
+    key = os.path.join(bucket_path, f'cache/{plugin}/{version}-manifest.json')
+    print(f'Processing {key}')
     # if the manifest for this plugin already exists there's nothing do to
     bucket = s3.Bucket(bucket_name)
-    existing_manifest_summary = bucket.objects.filter(Prefix=key)
+    existing_manifest_summary = list(bucket.objects.filter(Prefix=key))
+    print(f'Matching manifests in bucket: {existing_manifest_summary}')
     if existing_manifest_summary:
+        print("Manifest exists... returning.")
         return
 
+    # write file to s3 to ensure we never retry this plugin version
+    bucket.put_object(Body=json.dumps({}), Key=key)
     try:
+        print('Discovering manifest...')
         manifest = fetch_manifest(plugin, version)
         s3_body = manifest.json()
     except Exception as e:
+        print("Failed discovery...")
         s3_body =  json.dumps({'error': str(e)})
-    s3.put_object(Body=s3_body, Bucket=bucket_name, Key=key)
+    print(f'Writing {s3_body} to {key} in {bucket_name}')
+    bucket.put_object(Body=s3_body, Key=key)