kestra-io / plugin-gcp Goto Github PK

View Code? Open in Web Editor NEW

6.0 6.0 9.0 1.37 MB

License: Apache License 2.0

Java 99.88% R 0.04% Python 0.08%

google-cloud-platform google-cloud gcp kestra bigquery google-cloud-storage vertex-ai plugin firestore

plugin-gcp's Introduction

Event-Driven Declarative Orchestrator

Kestra infinitely scalable orchestration and scheduling platform

Get started with Kestra in 4 minutes.

Kestra Google Cloud Platform Plugin

Plugin to interract with Google Cloud Platform (GCP)

Documentation

Full documentation can be found under kestra.io/docs
Documentation for developing a plugin is included in the Plugin Developer Guide.

License

Apache 2.0 © Kestra Technologies

Stay up to date

We release new versions every month. Give the main repository a star to stay up to date with the latest releases and get notified about future updates.

plugin-gcp's People

Contributors

Stargazers

Watchers

Forkers

lucasdeleforge skraye asamoal aballiet gorgias shrutimantri aurelienwls inikitagricenko japerry911

plugin-gcp's Issues

PubSub Consume task fails with NullPointerException when no subscription is provided

Expected Behavior

When no subscription is provided, PubSub Consume task should automatically create a subscription as specified in the description of the subscription attribute.

Actual Behaviour

When no subscription is provided, PubSub Consume task fails with NullPointerException.

Here is the stacktrace:

2024-03-23 15:59:18.522 java.lang.NullPointerException
	at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:903)
	at com.google.pubsub.v1.ProjectSubscriptionName.<init>(ProjectSubscriptionName.java:57)
	at com.google.pubsub.v1.ProjectSubscriptionName.<init>(ProjectSubscriptionName.java:29)
	at com.google.pubsub.v1.ProjectSubscriptionName$Builder.build(ProjectSubscriptionName.java:157)
	at com.google.pubsub.v1.ProjectSubscriptionName.of(ProjectSubscriptionName.java:61)
	at io.kestra.plugin.gcp.pubsub.AbstractPubSub.createSubscription(AbstractPubSub.java:45)
	at io.kestra.plugin.gcp.pubsub.Consume.run(Consume.java:84)
	at io.kestra.plugin.gcp.pubsub.Consume.run(Consume.java:29)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:710)

Steps To Reproduce

Take the example flow, and put in appropriate values.
Do not add the subscription attribute to the task.
Execute the task. The execution fails with NullPointerException.

I also tried by adding autoCreateSubscription: true, but still the same error.

Environment Information

Kestra Version: 0.15.5
Plugin version: 0.15.5
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker): N/A

Example flow

id: pubsub-consume
namespace: dev
tasks:
  - id: "consume"
    type: "io.kestra.plugin.gcp.pubsub.Consume"
    projectId: <project-id>
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    topic: smantri-test
    maxRecords: 3

Results downloaded by Pubsub Consume task cannot be previewed

Expected Behavior

Results downloaded by Pubsub Consume task can be previewed.

Actual Behaviour

Preview on the downloaded results from PubSub Consume task failed to be previewed. It gives Error 500.

However, it is able to read the uri using Render Expression.

Steps To Reproduce

Use the example Publish task to publish the messages.
Then run the Consume task to consume these messages.
Preview the output of the Consume task.

Environment Information

Kestra Version: 0.15.5
Plugin version: 0.15.5
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker): N/A

Example flow

Publish Task:

id: "publish"
type: "io.kestra.plugin.gcp.pubsub.Publish"
projectId: <project-id>
serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
topic: topic-test
from:
-  data: "{{ 'base64-encoded-string-1' | base64encode }}"
   attributes:
       testAttribute: KestraTest
   messageId: '1234'
-  data: "{{ 'base64-encoded-string-2' | base64encode }}"
-  attributes:
       testAttribute: KestraTest

Consume Task

id: pubsub-consume
namespace: dev
tasks:
  - id: "consume"
    type: "io.kestra.plugin.gcp.pubsub.Consume"
    projectId: <project-id>
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    topic: smantri-test
    maxRecords: 3
    subscription: <subscription-id>

GCS Copy unable to copy from local storage files

Expected Behavior

Given a file to be copied from local storage to GCS, the file should get copied to the appropriate GCS bucket destination successfully.
Note: The Copy task works well for copying files from one GCS location into another, but is unable to copy from local storage into GCS bucket.

Actual Behaviour

As per the description of this task on this page, the file present in Kestra's internal storage can be copied to GCS location.

But I get the following error while executing the flow with the Copy task:

2024-01-29 15:23:23.834 • Using service account: kestra-gcs-svc-account@shruti-demo-project.iam.gserviceaccount.com
2024-01-29 15:23:23.837Moving from 'kestra:///company/team/gcp-gcs-copy/executions/63mIJjgxO0uzupqdl5mRdr/tasks/http-download/4ZediJHm6aN6YgMZCNx4Fm/7040886805462740981.csv' to 'gs://smantri-kestra-bucket/test/salaries/salaries.csv'
2024-01-29 15:23:25.178The billing account for the owning project is disabled in state closed
2024-01-29 15:23:25.178403 Forbidden
POST https://storage.googleapis.com/storage/v1/b/kestra/o/company%2Fteam%2Fgcp-gcs-copy%2Fexecutions%2F63mIJjgxO0uzupqdl5mRdr%2Ftasks%2Fhttp-download%2F4ZediJHm6aN6YgMZCNx4Fm%2F7040886805462740981.csv/rewriteTo/b/smantri-kestra-bucket/o/test%2Fsalaries%2Fsalaries.csv?projection=full
{
  "code" : 403,
  "errors" : [ {
    "domain" : "global",
    "location" : "Authorization",
    "locationType" : "header",
    "message" : "The billing account for the owning project is disabled in state closed",
    "reason" : "accountDisabled"
  } ],
  "message" : "The billing account for the owning project is disabled in state closed"
}
2024-01-29 15:23:25.178com.google.cloud.storage.StorageException: The billing account for the owning project is disabled in state closed
	at com.google.cloud.storage.StorageException.translate(StorageException.java:165)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.translate(HttpStorageRpc.java:313)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.rewrite(HttpStorageRpc.java:1162)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.openRewrite(HttpStorageRpc.java:1090)
	at com.google.cloud.storage.StorageImpl.lambda$copy$18(StorageImpl.java:593)
	at com.google.api.gax.retrying.DirectRetryingExecutor.submit(DirectRetryingExecutor.java:103)
	at com.google.cloud.RetryHelper.run(RetryHelper.java:76)
	at com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:50)
	at com.google.cloud.storage.Retrying.run(Retrying.java:65)
	at com.google.cloud.storage.StorageImpl.run(StorageImpl.java:1515)
	at com.google.cloud.storage.StorageImpl.copy(StorageImpl.java:591)
	at io.kestra.plugin.gcp.gcs.Copy.run(Copy.java:77)
	at io.kestra.plugin.gcp.gcs.Copy.run(Copy.java:21)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:684)
Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 403 Forbidden
POST https://storage.googleapis.com/storage/v1/b/kestra/o/company%2Fteam%2Fgcp-gcs-copy%2Fexecutions%2F63mIJjgxO0uzupqdl5mRdr%2Ftasks%2Fhttp-download%2F4ZediJHm6aN6YgMZCNx4Fm%2F7040886805462740981.csv/rewriteTo/b/smantri-kestra-bucket/o/test%2Fsalaries%2Fsalaries.csv?projection=full
{
  "code" : 403,
  "errors" : [ {
    "domain" : "global",
    "location" : "Authorization",
    "locationType" : "header",
    "message" : "The billing account for the owning project is disabled in state closed",
    "reason" : "accountDisabled"
  } ],
  "message" : "The billing account for the owning project is disabled in state closed"
}
	at com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:146)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:118)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:37)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest$3.interceptResponse(AbstractGoogleClientRequest.java:466)
	at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1111)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:552)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:493)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:603)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.rewrite(HttpStorageRpc.java:1152)
	... 11 more

Steps To Reproduce

Create a flow.
First task in the flow can be to download a file from http. This downloaded file can be used to copy to GCS.
Second task will be the Copy task with from being the internal storage file that has got downloaded from the first task. You can copy this file to any GCS location.
Execute the flow.

Environment Information

Kestra Version: 0.13.8
Plugin version: 0.13.8
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker):

Example flow

id: gcp-gcs-copy
namespace: company.team
tasks:
  - id: http_download
    type: io.kestra.plugin.fs.http.Download
    uri: https://raw.githubusercontent.com/kestra-io/datasets/main/csv/salaries.csv
  - id: "copy"
    type: "io.kestra.plugin.gcp.gcs.Copy"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    projectId: "<project_namw>"
    from: "{{ outputs.http_download.uri }}"
    to: "gs://<bucket_name>/test/salaries/salaries.csv"

Add `database` property to the GCP Firestore task

Feature description

GCP Firestore tasks do not have "database" as input. As a result, they only support (default) database.
Any production use-case would have some meaningful name for the database, and hence, our task should start accepting database name in the task attributes.

GCS DeleteBucket expects the bucket to be empty to be deleted

Expected Behavior

The given bucket should be deleted. We can provide an option to delete non-empty bucket, or explicitly mention that only empty buckets will be deleted.

Actual Behaviour

The non-empty bucket does not get deleted. The task fails with the following error:

2024-01-29 16:09:11.283Deleting bucket 'smantri-gcs-kestra-test-bucket'
2024-01-29 16:09:12.910The bucket you tried to delete is not empty.
2024-01-29 16:09:12.910409 Conflict
DELETE https://storage.googleapis.com/storage/v1/b/smantri-gcs-kestra-test-bucket
{
  "code" : 409,
  "errors" : [ {
    "domain" : "global",
    "message" : "The bucket you tried to delete is not empty.",
    "reason" : "conflict"
  } ],
  "message" : "The bucket you tried to delete is not empty."
}
2024-01-29 16:09:12.910com.google.cloud.storage.StorageException: The bucket you tried to delete is not empty.
	at com.google.cloud.storage.StorageException.translate(StorageException.java:165)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.translate(HttpStorageRpc.java:313)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.delete(HttpStorageRpc.java:642)
	at com.google.cloud.storage.StorageImpl.lambda$delete$14(StorageImpl.java:516)
	at com.google.api.gax.retrying.DirectRetryingExecutor.submit(DirectRetryingExecutor.java:103)
	at com.google.cloud.RetryHelper.run(RetryHelper.java:76)
	at com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:50)
	at com.google.cloud.storage.Retrying.run(Retrying.java:65)
	at com.google.cloud.storage.StorageImpl.run(StorageImpl.java:1515)
	at com.google.cloud.storage.StorageImpl.delete(StorageImpl.java:516)
	at io.kestra.plugin.gcp.gcs.DeleteBucket.run(DeleteBucket.java:54)
	at io.kestra.plugin.gcp.gcs.DeleteBucket.run(DeleteBucket.java:19)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:684)
Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 409 Conflict
DELETE https://storage.googleapis.com/storage/v1/b/smantri-gcs-kestra-test-bucket
{
  "code" : 409,
  "errors" : [ {
    "domain" : "global",
    "message" : "The bucket you tried to delete is not empty.",
    "reason" : "conflict"
  } ],
  "message" : "The bucket you tried to delete is not empty."
}
	at com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:146)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:118)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:37)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest$3.interceptResponse(AbstractGoogleClientRequest.java:466)
	at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1111)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:552)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:493)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:603)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.delete(HttpStorageRpc.java:638)
	... 10 more

Steps To Reproduce

in the DeleteBucket task, provide a non0-empty bucket to be deleted.

Environment Information

Kestra Version: 0.13.8
Plugin version: 0.13.8
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker):

Example flow

id: gcp-gcs-delete-bucket
namespace: company.team
tasks:
  - id: "delete"
    type: "io.kestra.plugin.gcp.gcs.DeleteBucket"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    projectId: "<project-name>"
    name: "<bucket-name>"

GKE ClusterMetadata task support for regional clusters

Feature description

As mentioned in #335, GKE ClusterMetadata task supports only zonal clusters. We should provide support for regional clusters as well, as production clusters tend to be regional.

GCS Compose fails with an 404 error

Expected Behavior

Given a GCS directory path containing few files, the files should concatenate into one file, and the concatenated file should be uploaded to the destination.

Actual Behaviour

The flow execution results into an error:

2024-01-29 15:00:07.828 • Using service account: kestra-gcs-svc-account@shruti-demo-project.iam.gserviceaccount.com
2024-01-29 15:00:07.836 • Using service account: kestra-gcs-svc-account@shruti-demo-project.iam.gserviceaccount.com
2024-01-29 15:00:10.163Found '4' blobs from 'gs://smantri-gcs-kestra-test-bucket/test/users/'
2024-01-29 15:00:10.167Compose from '[com.google.cloud.storage.Storage$ComposeRequest$SourceBlob@3221112c, com.google.cloud.storage.Storage$ComposeRequest$SourceBlob@1afa53f0, com.google.cloud.storage.Storage$ComposeRequest$SourceBlob@672893f7, com.google.cloud.storage.Storage$ComposeRequest$SourceBlob@1fb84247]' to 'gs://smantri-kestra-bucket/test/users.csv'
2024-01-29 15:00:10.597Object test/users/ (generation: 0) not found.
2024-01-29 15:00:10.597404 Not Found
POST https://storage.googleapis.com/storage/v1/b/smantri-kestra-bucket/o/test%2Fusers.csv/compose
{
  "code" : 404,
  "errors" : [ {
    "domain" : "global",
    "message" : "Object test/users/ (generation: 0) not found.",
    "reason" : "notFound"
  } ],
  "message" : "Object test/users/ (generation: 0) not found."
}
2024-01-29 15:00:10.597com.google.cloud.storage.StorageException: Object test/users/ (generation: 0) not found.
	at com.google.cloud.storage.StorageException.translate(StorageException.java:165)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.translate(HttpStorageRpc.java:313)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.compose(HttpStorageRpc.java:716)
	at com.google.cloud.storage.StorageImpl.lambda$compose$16(StorageImpl.java:562)
	at com.google.api.gax.retrying.DirectRetryingExecutor.submit(DirectRetryingExecutor.java:103)
	at com.google.cloud.RetryHelper.run(RetryHelper.java:76)
	at com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:50)
	at com.google.cloud.storage.Retrying.run(Retrying.java:65)
	at com.google.cloud.storage.StorageImpl.run(StorageImpl.java:1515)
	at com.google.cloud.storage.StorageImpl.compose(StorageImpl.java:560)
	at io.kestra.plugin.gcp.gcs.Compose.run(Compose.java:102)
	at io.kestra.plugin.gcp.gcs.Compose.run(Compose.java:21)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:684)
Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 404 Not Found
POST https://storage.googleapis.com/storage/v1/b/smantri-kestra-bucket/o/test%2Fusers.csv/compose
{
  "code" : 404,
  "errors" : [ {
    "domain" : "global",
    "message" : "Object test/users/ (generation: 0) not found.",
    "reason" : "notFound"
  } ],
  "message" : "Object test/users/ (generation: 0) not found."
}
	at com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:146)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:118)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:37)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest$3.interceptResponse(AbstractGoogleClientRequest.java:466)
	at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1111)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:552)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:493)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:603)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.compose(HttpStorageRpc.java:713)
	... 10 more

Steps To Reproduce

Create a bucket, and in a particular folder in that bucket, create a few files that need to be concatenated.
Create a new bucket where the concatenated file will be uploaded.
Create a flow with the task as shown in the example on this page.
Execute the flow.

Environment Information

Kestra Version:
Plugin version:
Operating System (OS / Docker / Kubernetes):
Java Version (If not docker):

Example flow

id: gcp-gcs-compose
namespace: company.team
tasks:
  - id: "compose"
    type: "io.kestra.plugin.gcp.gcs.Compose"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    projectId: "<project_name>"
    list: 
      from: "gs://<bucket1>/test/users/"
    to: "gs://<bucket2>/test/users.csv"

This location: gs://<bucket1>/test/users/ contains 3 files: users1.csv, users2.csv, users3.csv

Dataproc job - mandatory projectId field

Expected Behavior

No response

Actual Behaviour

projectId is mandatory but not validated for the moment (so if the user don't provide the projectId it will end up with error)

id: spark_job
namespace: demo

tasks:

  - id: spark2
    type: io.kestra.plugin.gcp.dataproc.batches.PySparkSubmit
    fileUris: 
      - 'gs://bpimpaud-demo/spark/files'
    jarFileUris: 
      - 'gs://bpimpaud-demo/spark/jar'
    mainPythonFileUri: 'gs://bpimpaud-demo/spark/main.py'
    region: EU
    archiveUris: 
      - 'gs://bpimpaud-demo/spark/archive'
    name: test-pyspark
    args:
      - 'hello'

Steps To Reproduce

No response

Environment Information

Kestra Version:
Plugin version:
Operating System (OS / Docker / Kubernetes):
Java Version (If not docker):

Example flow

No response

Bigquery Read Timeout should retry

Seems that this error is not retry by google cloud Java library, need to dig to find the reasons

Dataflow

Spanner

https://cloud.google.com/spanner

GCS CreateBucketIAMPolicy task example not working as expected

Expected Behavior

The role should get assigned to the member for the particular bucket present on the task.

Actual Behaviour

Getting the following error on executing the flow with the example task on this page:

PUT https://storage.googleapis.com/storage/v1/b/smantri-gcs-kestra-test-bucket/iam
{
  "code" : 400,
  "errors" : [ {
    "domain" : "global",
    "message" : "The member [email protected] is of an unknown type. Please set a valid type prefix for the member.",
    "reason" : "invalid"
  } ],
  "message" : "The member [email protected] is of an unknown type. Please set a valid type prefix for the member."
}
2024-01-29 14:16:20.781com.google.cloud.storage.StorageException: The member [email protected] is of an unknown type. Please set a valid type prefix for the member.
	at com.google.cloud.storage.StorageException.translate(StorageException.java:165)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.translate(HttpStorageRpc.java:313)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.setIamPolicy(HttpStorageRpc.java:1627)
	at com.google.cloud.storage.StorageImpl.lambda$setIamPolicy$48(StorageImpl.java:1469)
	at com.google.api.gax.retrying.DirectRetryingExecutor.submit(DirectRetryingExecutor.java:103)
	at com.google.cloud.RetryHelper.run(RetryHelper.java:76)
	at com.google.cloud.RetryHelper.runWithRetries(RetryHelper.java:50)
	at com.google.cloud.storage.Retrying.run(Retrying.java:65)
	at com.google.cloud.storage.StorageImpl.run(StorageImpl.java:1515)
	at com.google.cloud.storage.StorageImpl.setIamPolicy(StorageImpl.java:1467)
	at io.kestra.plugin.gcp.gcs.CreateBucketIamPolicy.run(CreateBucketIamPolicy.java:112)
	at io.kestra.plugin.gcp.gcs.CreateBucketIamPolicy.run(CreateBucketIamPolicy.java:20)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:684)
Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 400 Bad Request
PUT https://storage.googleapis.com/storage/v1/b/smantri-gcs-kestra-test-bucket/iam
{
  "code" : 400,
  "errors" : [ {
    "domain" : "global",
    "message" : "The member [email protected] is of an unknown type. Please set a valid type prefix for the member.",
    "reason" : "invalid"
  } ],
  "message" : "The member [email protected] is of an unknown type. Please set a valid type prefix for the member."
}
	at com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:146)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:118)
	at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:37)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest$3.interceptResponse(AbstractGoogleClientRequest.java:466)
	at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1111)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:552)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:493)
	at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:603)
	at com.google.cloud.storage.spi.v1.HttpStorageRpc.setIamPolicy(HttpStorageRpc.java:1624)
	... 10 more

Steps To Reproduce

I have provided "Security Admin" and "Storage Admin" roles to the service account used in "serviceAccount" property.
Created another service account: "[email protected]", and have not provided it any roles.
Used the flow as mentioned in the example task on this page

Environment Information

Kestra Version:
Plugin version:
Operating System (OS / Docker / Kubernetes):
Java Version (If not docker):

Example flow

id: gcp-gcs-create-bucket-iam-policy
namespace: company.team
tasks:
  - id: "create_bucket_iam_policy"
    type: "io.kestra.plugin.gcp.gcs.CreateBucketIamPolicy"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    projectId: "<project_name>"
    name: "<bucket_name>"
    member: "<new_service_account>"
    role: "roles/storage.admin"

io.kestra.plugin.gcp.gcs.Upload allow setting object content-type

Feature description

I have a use case where I want to put images into a bucket and have them viewable (not downloaded) through a browser. This requires setting an appropriate content-type on the placed object. I can do this with the gcloud sdk in a container, but it would be nice to use io.kestra.plugin.gcp.gcs.Upload to do the upload with the content-type.

Metrics are not rendered

I get stuff like

2020-01-29 17:11:24,162 TRACE mory-queue-0 .bq-load-stg Metrics
- name: output.rows
  tags:
    destination_table: '{{ globals.projects-stg }}.samples.{{ flow.id }}_v1_{{ execution.id
      }}'
    projectId: my-project-dev
    location: EU

Output in my Metrics. it would be nice if it was rendered

Pub/Sub Lite

Feature description

Provides support for Pub/Sub Lite which uses a different API than Pub/Sub, see https://github.com/googleapis/java-pubsublite

Talend

Secrets

Feature description

Currently in order to retrieve secrets hosted on GCP, we need to perform command :

gcloud secrets versions access "latest" --secret "MY_SECRET" using the CLI.

Current flow looks like this :

id: retrieve-gcloud-secret
namespace: dev
tasks:
  - id: get-credentials
    type: "io.kestra.plugin.gcp.cli.GCloudCLI"
    serviceAccount: "{{ globals['my-sa-json'] }}"
    commands:
      - gcloud secrets versions access "latest" --secret "MY_SECRETS_ON_GCP" --format json | tr -d '\n ' | xargs -0 -I {} echo '::{"outputs":{"gcloud":{}}}::'
  - id: hello
    type: io.kestra.core.tasks.log.Log
    message: "Your secret value : {{ outputs['get-credentials'].vars['gcloud']['payload']['data'] }}"

Aside that the secret value is displayed in the logs, could be interesting to access them more easily

I am willing to do a PR 👍

GKE ClusterMetadata task failing to run successfully

Expected Behavior

The GKE ClusterMetadata task should be able to get the cluster's metadata.

Actual Behaviour

I am using the following flow by going through the ClusterMetadata task page:

id: gke-cluster-metadata
namespace: dev
tasks:
  - id: "cluster_metadata"
    type: "io.kestra.plugin.gcp.gke.ClusterMetadata"
    clusterId: "my-kestra-cluster"
    clusterProjectId: "<GCP-project-id>"
    clusterZone: "europe-west3"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"

The flow fails with the following error:

2024-03-17 19:25:13.802 • Using service account: smantri-kestra-20230317@empirical-realm-374307.iam.gserviceaccount.com
2024-03-17 19:25:14.595io.grpc.StatusRuntimeException: INVALID_ARGUMENT: 'zone' field cannot be used to access GKE regional clusters. Use 'name' field instead.
2024-03-17 19:25:14.595INVALID_ARGUMENT: 'zone' field cannot be used to access GKE regional clusters. Use 'name' field instead.
2024-03-17 19:25:14.595com.google.api.gax.rpc.InvalidArgumentException: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: 'zone' field cannot be used to access GKE regional clusters. Use 'name' field instead.
	at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:92)
	at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:98)
	at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:66)
	at com.google.api.gax.grpc.GrpcExceptionCallable$ExceptionTransformingFuture.onFailure(GrpcExceptionCallable.java:97)
	at com.google.api.core.ApiFutures$1.onFailure(ApiFutures.java:84)
	at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1128)
	at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:31)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1298)
	at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:1059)
	at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:809)
	at io.grpc.stub.ClientCalls$GrpcFuture.setException(ClientCalls.java:568)
	at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:538)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:570)
	at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:489)
	at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:453)
	at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:486)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:574)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:72)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:742)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:723)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
	Suppressed: com.google.api.gax.rpc.AsyncTaskException: Asynchronous task failed
		at com.google.api.gax.rpc.ApiExceptions.callAndTranslateApiException(ApiExceptions.java:57)
		at com.google.api.gax.rpc.UnaryCallable.call(UnaryCallable.java:112)
		at com.google.cloud.container.v1.ClusterManagerClient.getCluster(ClusterManagerClient.java:1022)
		at com.google.cloud.container.v1.ClusterManagerClient.getCluster(ClusterManagerClient.java:991)
		at io.kestra.plugin.gcp.gke.ClusterMetadata.fetch(ClusterMetadata.java:107)
		at io.kestra.plugin.gcp.gke.ClusterMetadata.run(ClusterMetadata.java:66)
		at io.kestra.plugin.gcp.gke.ClusterMetadata.run(ClusterMetadata.java:23)
		at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:710)
Caused by: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: 'zone' field cannot be used to access GKE regional clusters. Use 'name' field instead.
	at io.grpc.Status.asRuntimeException(Status.java:533)
	... 17 more

Steps To Reproduce

Use the flow as the one in Example flow section below.
Run the flow. The execution fails with an error.

Environment Information

Kestra Version: 0.15.0
Plugin version: 0.15.0
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker):

Example flow

id: gke-cluster-metadata
namespace: dev
tasks:
  - id: "cluster_metadata"
    type: "io.kestra.plugin.gcp.gke.ClusterMetadata"
    clusterId: "my-kestra-cluster"
    clusterProjectId: "<GCP-project-id>"
    clusterZone: "europe-west3"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"

Auth to Google Artifact Registry fails even when using `OauthAccessToken`

Expected Behavior

Example flow:

id: gcpArtifactRegistry
namespace: dev
tasks:
  - id: setup
    type: io.kestra.core.tasks.flows.Worker
    tasks:
      - id: cloneRepository
        type: io.kestra.plugin.git.Clone
        branch: main
        url: https://github.com/anna-geller/kestra-flows
      
      - id: fetchAuthToken
        type: io.kestra.plugin.gcp.auth.OauthAccessToken
        projectId: geller
        serviceAccount: "{{envs.gcp_creds}}"

      - id: pythonTask
        type: io.kestra.core.tasks.scripts.Bash
        commands:
          - python python-scripts/etl_script.py
        dockerOptions:
          image: europe-west3-docker.pkg.dev/geller/flows/python:latest
          dockerConfig: |
            {
              "auths": {
                "europe-west3-docker.pkg.dev": {
                    "username": "_json_keyfile",
                    "password": "{{outputs.fetchAuthToken.accessToken.tokenValue}}"
                    }
              },
              "credHelpers": {
                "europe-west3-docker.pkg.dev": "gcloud"
              }
            }
        runner: DOCKER

Actual Behaviour

Without running one of these commands (either gcloud auth or docker login) before, it doesn't seem to work:

gcloud auth configure-docker europe-west3-docker.pkg.dev
cat mykey.json | docker login -u _json_key --password-stdin https://europe-west3-docker.pkg.dev

It might be required to either:

add the gcloud CLI plugin https://github.com/orgs/kestra-io/projects/3/views/1?filterQuery=gcloud&pane=issue&itemId=27437640
add the Docker plugin https://github.com/orgs/kestra-io/projects/3/views/1?pane=issue&itemId=30479463

To support it?

Environment Information

Kestra Version: develop-full docker 0.10.0

Add a task to invoke a Google Cloud Function

Feature description

Add support for invoking GCP Cloud Functions - it would be great to allow providing custom inputs to the function, invoking it, waiting for its completion, and fetching the response payload to outputs, equivalent to Lambda Invoke task on AWS implementation

Dataproc serverless

Firestore

Support BigQuery JSON columns

Feature description

We'd like to "select" JSON columns from BigQuery into the output. Currently when attempting a simple select that contains JSON column, we get the following error:

Example select:

select data from my_table limit 1

Error:

Invalid type 'JSON'

Example task:

  - id: "query"
    type: "io.kestra.plugin.gcp.bigquery.Query"
    projectId: my-project
    serviceAccount: "{{secret('GCP_SERVICE_JSON')}}"
    fetchOne: true
    sql: |
      select data from my_table limit 1

Add validation on BigQuery Load

when filetype is CSV, csvOptions is mandatory, add a control on it

VertexAI MultimodalCompletion task fails with an error

Expected Behavior

VertexAI MultimodalCompletion task should run successfully.

Actual Behaviour

VertexAI Multimodal task fails with the following error:

2024-03-17 22:55:12.552 • Using service account: smantri-kestra-20230317@empirical-realm-374307.iam.gserviceaccount.com
2024-03-17 22:55:13.083io.grpc.StatusRuntimeException: UNIMPLEMENTED: HTTP status code 404
invalid content-type: text/html; charset=UTF-8
headers: Metadata(:status=404,content-type=text/html; charset=UTF-8,referrer-policy=no-referrer,date=Sun, 17 Mar 2024 17:25:13 GMT,alt-svc=h3=":443"; ma=2592000,h3-29=":443"; ma=2592000,content-length=1627)
DATA-----------------------------
<!DOCTYPE html>
<html lang=en>
  <meta charset=utf-8>
  <meta name=viewport content="initial-scale=1, minimum-scale=1, width=device-width">
  <title>Error 404 (Not Found)!!1</title>
  <style>
    *{margin:0;padding:0}html,code{font:15px/22px arial,sans-serif}html{background:#fff;color:#222;padding:15px}body{margin:7% auto 0;max-width:390px;min-height:180px;padding:30px 0 15px}* > body{background:url(//www.google.com/images/errors/robot.png) 100% 5px no-repeat;padding-right:205px}p{margin:11px 0 22px;overflow:hidden}ins{color:#777;text-decoration:none}a img{border:0}@media screen and (max-width:772px){body{background:none;margin-top:0;max-width:none;padding-right:0}}#logo{background:url(//www.google.com/images/branding/googlelogo/1x/googlelogo_color_150x54dp.png) no-repeat;margin-left:-5px}@media only screen and (min-resolution:192dpi){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat 0% 0%/100% 100%;-moz-border-image:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) 0}}@media only screen and (-webkit-min-device-pixel-ratio:2){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat;-webkit-background-size:100% 100%}}#logo{display:inline-block;height:54px;width:150px}
  </style>
  <a href=//www.google.com/></a>
  <p><b>404.</b> <ins>That’s an error.</ins>
  <p>The requested URL <code>/google.cloud.aiplatform.v1.PredictionService/StreamGenerateContent</code> was not found on this server.  <ins>That’s all we know.</ins>
2024-03-17 22:55:13.083UNIMPLEMENTED: HTTP status code 404
invalid content-type: text/html; charset=UTF-8
headers: Metadata(:status=404,content-type=text/html; charset=UTF-8,referrer-policy=no-referrer,date=Sun, 17 Mar 2024 17:25:13 GMT,alt-svc=h3=":443"; ma=2592000,h3-29=":443"; ma=2592000,content-length=1627)
DATA-----------------------------
<!DOCTYPE html>
<html lang=en>
  <meta charset=utf-8>
  <meta name=viewport content="initial-scale=1, minimum-scale=1, width=device-width">
  <title>Error 404 (Not Found)!!1</title>
  <style>
    *{margin:0;padding:0}html,code{font:15px/22px arial,sans-serif}html{background:#fff;color:#222;padding:15px}body{margin:7% auto 0;max-width:390px;min-height:180px;padding:30px 0 15px}* > body{background:url(//www.google.com/images/errors/robot.png) 100% 5px no-repeat;padding-right:205px}p{margin:11px 0 22px;overflow:hidden}ins{color:#777;text-decoration:none}a img{border:0}@media screen and (max-width:772px){body{background:none;margin-top:0;max-width:none;padding-right:0}}#logo{background:url(//www.google.com/images/branding/googlelogo/1x/googlelogo_color_150x54dp.png) no-repeat;margin-left:-5px}@media only screen and (min-resolution:192dpi){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat 0% 0%/100% 100%;-moz-border-image:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) 0}}@media only screen and (-webkit-min-device-pixel-ratio:2){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat;-webkit-background-size:100% 100%}}#logo{display:inline-block;height:54px;width:150px}
  </style>
  <a href=//www.google.com/></a>
  <p><b>404.</b> <ins>That’s an error.</ins>
  <p>The requested URL <code>/google.cloud.aiplatform.v1.PredictionService/StreamGenerateContent</code> was not found on this server.  <ins>That’s all we know.</ins>
2024-03-17 22:55:13.083com.google.api.gax.rpc.UnimplementedException: io.grpc.StatusRuntimeException: UNIMPLEMENTED: HTTP status code 404
invalid content-type: text/html; charset=UTF-8
headers: Metadata(:status=404,content-type=text/html; charset=UTF-8,referrer-policy=no-referrer,date=Sun, 17 Mar 2024 17:25:13 GMT,alt-svc=h3=":443"; ma=2592000,h3-29=":443"; ma=2592000,content-length=1627)
DATA-----------------------------
<!DOCTYPE html>
<html lang=en>
  <meta charset=utf-8>
  <meta name=viewport content="initial-scale=1, minimum-scale=1, width=device-width">
  <title>Error 404 (Not Found)!!1</title>
  <style>
    *{margin:0;padding:0}html,code{font:15px/22px arial,sans-serif}html{background:#fff;color:#222;padding:15px}body{margin:7% auto 0;max-width:390px;min-height:180px;padding:30px 0 15px}* > body{background:url(//www.google.com/images/errors/robot.png) 100% 5px no-repeat;padding-right:205px}p{margin:11px 0 22px;overflow:hidden}ins{color:#777;text-decoration:none}a img{border:0}@media screen and (max-width:772px){body{background:none;margin-top:0;max-width:none;padding-right:0}}#logo{background:url(//www.google.com/images/branding/googlelogo/1x/googlelogo_color_150x54dp.png) no-repeat;margin-left:-5px}@media only screen and (min-resolution:192dpi){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat 0% 0%/100% 100%;-moz-border-image:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) 0}}@media only screen and (-webkit-min-device-pixel-ratio:2){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat;-webkit-background-size:100% 100%}}#logo{display:inline-block;height:54px;width:150px}
  </style>
  <a href=//www.google.com/></a>
  <p><b>404.</b> <ins>That’s an error.</ins>
  <p>The requested URL <code>/google.cloud.aiplatform.v1.PredictionService/StreamGenerateContent</code> was not found on this server.  <ins>That’s all we know.</ins>

	at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:108)
	at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:41)
	at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:86)
	at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:66)
	at com.google.api.gax.grpc.ExceptionResponseObserver.onErrorImpl(ExceptionResponseObserver.java:82)
	at com.google.api.gax.rpc.StateCheckingResponseObserver.onError(StateCheckingResponseObserver.java:84)
	at com.google.api.gax.grpc.GrpcDirectStreamController$ResponseObserverAdapter.onClose(GrpcDirectStreamController.java:148)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:570)
	at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:489)
	at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:453)
	at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:486)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:574)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:72)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:742)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:723)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
	Suppressed: java.lang.RuntimeException: Asynchronous task failed
		at com.google.api.gax.rpc.ServerStreamIterator.hasNext(ServerStreamIterator.java:105)
		at com.google.cloud.vertexai.generativeai.ResponseStreamIteratorWithHistory.hasNext(ResponseStreamIteratorWithHistory.java:37)
		at com.google.cloud.vertexai.generativeai.ResponseHandler.aggregateStreamIntoResponse(ResponseHandler.java:104)
		at com.google.cloud.vertexai.generativeai.GenerativeModel.generateContent(GenerativeModel.java:547)
		at com.google.cloud.vertexai.generativeai.GenerativeModel.generateContent(GenerativeModel.java:612)
		at com.google.cloud.vertexai.generativeai.GenerativeModel.generateContent(GenerativeModel.java:560)
		at io.kestra.plugin.gcp.vertexai.MultimodalCompletion.run(MultimodalCompletion.java:95)
		at io.kestra.plugin.gcp.vertexai.MultimodalCompletion.run(MultimodalCompletion.java:33)
		at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:710)
Caused by: io.grpc.StatusRuntimeException: UNIMPLEMENTED: HTTP status code 404
invalid content-type: text/html; charset=UTF-8
headers: Metadata(:status=404,content-type=text/html; charset=UTF-8,referrer-policy=no-referrer,date=Sun, 17 Mar 2024 17:25:13 GMT,alt-svc=h3=":443"; ma=2592000,h3-29=":443"; ma=2592000,content-length=1627)
DATA-----------------------------
<!DOCTYPE html>
<html lang=en>
  <meta charset=utf-8>
  <meta name=viewport content="initial-scale=1, minimum-scale=1, width=device-width">
  <title>Error 404 (Not Found)!!1</title>
  <style>
    *{margin:0;padding:0}html,code{font:15px/22px arial,sans-serif}html{background:#fff;color:#222;padding:15px}body{margin:7% auto 0;max-width:390px;min-height:180px;padding:30px 0 15px}* > body{background:url(//www.google.com/images/errors/robot.png) 100% 5px no-repeat;padding-right:205px}p{margin:11px 0 22px;overflow:hidden}ins{color:#777;text-decoration:none}a img{border:0}@media screen and (max-width:772px){body{background:none;margin-top:0;max-width:none;padding-right:0}}#logo{background:url(//www.google.com/images/branding/googlelogo/1x/googlelogo_color_150x54dp.png) no-repeat;margin-left:-5px}@media only screen and (min-resolution:192dpi){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat 0% 0%/100% 100%;-moz-border-image:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) 0}}@media only screen and (-webkit-min-device-pixel-ratio:2){#logo{background:url(//www.google.com/images/branding/googlelogo/2x/googlelogo_color_150x54dp.png) no-repeat;-webkit-background-size:100% 100%}}#logo{display:inline-block;height:54px;width:150px}
  </style>
  <a href=//www.google.com/></a>
  <p><b>404.</b> <ins>That’s an error.</ins>
  <p>The requested URL <code>/google.cloud.aiplatform.v1.PredictionService/StreamGenerateContent</code> was not found on this server.  <ins>That’s all we know.</ins>

	at io.grpc.Status.asRuntimeException(Status.java:533)
	... 17 more

I am not sure if I am missing something here.

Steps To Reproduce

Execute the flow in the sample example with appropriate values.
The execution fails with an error.

Environment Information

Kestra Version: 0.15.0
Plugin version: 0.15.0
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker):

Example flow

id: vertexai-multimodal-completion
namespace: dev
tasks:
  - id: "multimodal_completion"
    type: "io.kestra.plugin.gcp.vertexai.MultimodalCompletion"
    region: "eu-west3"
    projectId: "<project-id>"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    contents:
      - content: Please tell me a joke

Worker error with vertexai Customjob tasks

Expected Behavior

No response

Actual Behaviour

Cloud logging shows a recurrent error when some vertexai tasks are over:

"message": "~* Previous channel ManagedChannelImpl{logId=270, target=logging.googleapis.com:443} was not shutdown properly!!! *~\n Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true.",
"context": "default",
"thread": "WorkerThread",
"logger": "io.grpc.internal.ManagedChannelOrphanWrapper"

Steps To Reproduce

No response

Environment Information

Kestra Version: 0.9.8
Plugin version: 0.9+

Example flow

tasks:

type: io.kestra.plugin.gcp.vertexai.CustomJob

Add Kestra info into BigQuery label

Feature description

When we run a BigQuery Query, it could be usefull to automatically add the Kestra namespace, flow, execution ID and task ID to the bigquery label

https://cloud.google.com/bigquery/docs/adding-labels#job-label

It's usefull to analyse into GCP billing dashboard which Kestra namespace / flow is the most expensive

PubSub Publish task is successful even when no PubSub topic exists

Expected Behavior

The PubSub Publish task should fail if the publish did not take place (when the topic itself did not exist).

Actual Behaviour

The PubSub Publish task is successful when there is no PubSub topic.

Steps To Reproduce

Ensure the PubSub topic does not exist.
Run the example flow of PubSub publish task.
Execution will be successful.

Environment Information

Kestra Version: 0.15.5
Plugin version: 0.15.5
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker):

Example flow

id: pubsub-publish
namespace: dev
tasks:
  - id: "publish"
    type: "io.kestra.plugin.gcp.pubsub.Publish"
    projectId: <project-id>
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    topic: smantri-test
    from:
    -  data: "{{ 'base64-encoded-string-1' | base64encode }}"
       attributes:
           testAttribute: KestraTest
       messageId: '1234'
    -  data: "{{ 'base64-encoded-string-2' | base64encode }}"
    -  attributes:
           testAttribute: KestraTest

Bigquery.Query: Need more datetime precision in outputs

Issue description

Hello

Request for more precisions in outputs when doing datetimes based queries in Bigquery.
Actually we have something like "2023-10-06T16:23:00.846Z" in outputs but "2023-10-06 16:23:00.846647" in BQ

We also need microseconds to do some strict comparisons between dates.
Thanks

Support JSON for Pub/Sub

Like others queue, we need to provide a SerdeType to deserialize JSON easily

Allow keeping original file format without ION serialization when downloading files from GCS to internal storage

Issue description

GCS Download task version of this issue kestra-io/kestra#1784

GCloud VertexAI Plugin broken with No JsonMapper implementation found

Expected Behavior

GCloud Vertex AI plugin task ChatCompletion should run successfully with the example given on the task page.

Actual Behaviour

GCloud Vertex AI plugin task ChatCompletion fails with the error:

2024-03-17 22:36:01.993 • Using service account: smantri-kestra-20230317@empirical-realm-374307.iam.gserviceaccount.com
2024-03-17 22:36:02.181Calling Vertex.AI prediction API https://eu-west3-aiplatform.googleapis.com/v1/projects/empirical-realm-374307/locations/eu-west3/publishers/google/models/chat-bison:predict
2024-03-17 22:36:02.245No JsonMapper implementation found
2024-03-17 22:36:02.245java.lang.IllegalStateException: No JsonMapper implementation found
	at io.micronaut.json.JsonMapper.lambda$createDefault$2(JsonMapper.java:367)
	at java.base/java.util.Optional.orElseThrow(Unknown Source)
	at io.micronaut.json.JsonMapper.createDefault(JsonMapper.java:367)
	at io.micronaut.http.client.netty.DefaultHttpClient.createDefaultMediaTypeRegistry(DefaultHttpClient.java:1796)
	at io.micronaut.http.client.netty.DefaultHttpClient.<init>(DefaultHttpClient.java:431)
	at io.micronaut.http.client.netty.DefaultHttpClient.<init>(DefaultHttpClient.java:417)
	at io.micronaut.http.client.netty.NettyHttpClientFactory.createNettyClient(NettyHttpClientFactory.java:132)
	at io.micronaut.http.client.netty.NettyHttpClientFactory.createNettyClient(NettyHttpClientFactory.java:121)
	at io.micronaut.http.client.netty.NettyHttpClientFactory.createClient(NettyHttpClientFactory.java:60)
	at io.kestra.plugin.gcp.vertexai.AbstractGenerativeAi.client(AbstractGenerativeAi.java:105)
	at io.kestra.plugin.gcp.vertexai.AbstractGenerativeAi.call(AbstractGenerativeAi.java:65)
	at io.kestra.plugin.gcp.vertexai.ChatCompletion.run(ChatCompletion.java:77)
	at io.kestra.plugin.gcp.vertexai.ChatCompletion.run(ChatCompletion.java:26)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:710)

Steps To Reproduce

Execute the flow in the Example flow by providing proper parameters.
The execution fails with an "No JsonMapper implementation found" error.

Environment Information

Kestra Version: 0.15.0
Plugin version: 0.15.0
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker):

Example flow

id: gcloud-vertexai-chat-completion
namespace: dev
tasks:
  - id: "chat_completion"
    type: "io.kestra.plugin.gcp.vertexai.ChatCompletion"
    region: eu-west3
    projectId: "<project-id>"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    context: I love jokes that talk about sport
    messages:
      - author: user
        content: Please tell me a joke

Allow the GCS file trigger to not download the file to internal storage

Feature description

Problem

GCS is an interesting edge case because the task LoadFromGcs doesn't require the trigger object to be downloaded to internal storage but only the file URL on GCS is needed

As @Ben8t mentioned, this would be useful for use cases where data is loaded from GCS to BigQuery based on GCS file detection event

Possible solutions

action: NONE (or sth like "do nothing") and if set to that value, not downloading the file to internal storage
expose the moved file to blobs property (atm we only get the original file information)

PubSub Publish task fails when message contains orderingKey

Expected Behavior

Message has an attribute orderingKey. Messages with orderingKey should be published.

Actual Behaviour

If the task has messages with orderingKey, the execution fails. Here is the error log:

2024-03-23 12:31:27.869 Cannot publish a message with an ordering key when message ordering is not enabled in the Publisher client. Please create a Publisher client with setEnableMessageOrdering(true) in the builder.
2024-03-23 12:31:27.869 java.lang.IllegalStateException: Cannot publish a message with an ordering key when message ordering is not enabled in the Publisher client. Please create a Publisher client with setEnableMessageOrdering(true) in the builder.
	at com.google.common.base.Preconditions.checkState(Preconditions.java:512)
	at com.google.cloud.pubsub.v1.Publisher.publish(Publisher.java:255)
	at io.kestra.plugin.gcp.pubsub.Publish.lambda$buildFlowable$1(Publish.java:116)
	at io.kestra.core.utils.Rethrow.lambda$throwFunction$4(Rethrow.java:90)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:113)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
	at reactor.core.publisher.FluxArray$ArraySubscription.fastPath(FluxArray.java:171)
	at reactor.core.publisher.FluxArray$ArraySubscription.request(FluxArray.java:96)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:171)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:171)
	at reactor.core.publisher.MonoReduce$ReduceSubscriber.request(MonoReduce.java:222)
	at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:53)
	at reactor.core.publisher.MonoReduce$ReduceSubscriber.onSubscribe(MonoReduce.java:98)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:96)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:96)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59)
	at reactor.core.publisher.Mono.subscribe(Mono.java:4512)
	at reactor.core.publisher.Mono.block(Mono.java:1727)
	at io.kestra.plugin.gcp.pubsub.Publish.run(Publish.java:97)
	at io.kestra.plugin.gcp.pubsub.Publish.run(Publish.java:28)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:710)
	Suppressed: java.lang.Exception: #block terminated with an error
		at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:103)
		at reactor.core.publisher.Mono.block(Mono.java:1728)
		... 3 more

Looks like we should enableMessageOrdering on the publisher when the messages contain orderingKey (or we can by default set it to enabled).

Steps To Reproduce

Take the example flow, and set correct values (ensure orderingKey is set for atleast one of the messages).
Execute the flow.

Environment Information

Kestra Version: 0.15.5
Plugin version: 0.15.5
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker): N/A

Example flow

id: pubsub-publish
namespace: dev
tasks:
  - id: "publish"
    type: "io.kestra.plugin.gcp.pubsub.Publish"
    projectId: <project-id>
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    topic: smantri-test
    from:
    -  data: "{{ 'base64-encoded-string-1' | base64encode }}"
       attributes:
           testAttribute: KestraTest
       messageId: '1234'
       orderingKey: 'foo'
    -  data: "{{ 'base64-encoded-string-2' | base64encode }}"
    -  attributes:
           testAttribute: KestraTest

BigTable

Feature description

BigTable

Gcs Storage File name is not an Java URI

Illegal character in path at index 63: gs://bucket/file_20210322 (1).csv
java.lang.IllegalArgumentException: Illegal character in path at index 63: gs://bucket/file_001_004_20210322 (1).csv
	at java.base/java.net.URI.create(Unknown Source)
	at org.kestra.task.gcp.gcs.models.Blob.of(Blob.java:50)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)

Maybe you must use a simple String ?

Pub/Sub

PubSub trigger doesn't fire

Issue description

Reproducer flow:

id: gcpPubSub
namespace: dev

tasks:
  - id: payload_log
    type: io.kestra.core.tasks.log.Log
    message: "{{ triggers.objects }}"

triggers:
  - id: get_payload
    type: io.kestra.plugin.gcp.pubsub.Trigger
    topic: kestra
    subscription: kestra-sub
    projectId: geller
    interval: PT1S
    maxRecords: 1
    serviceAccount: "{{ secret('GCP_CREDS') }}"
    autoCreateSubscription: false

Using a default setup, I created a topic "kestra" and a default subscription. Then I sent 2 messages from GCP console, yet no execution was triggered.

Prevent usage of store / fetch / fetchOne with destinationTable

We need to add a javax Validation to prevent the submission.
This case will lead at runtime to :
https://github.com/kestra-io/task-gcp/blob/20bb09bbe0b3c7bfd8862eda57bf9d92a596634e/src/main/java/org/kestra/task/gcp/bigquery/Query.java#L215

Delete GCS files according to a regex

Could you add an option to delete a list of files into a bucket that match a regex

use AbstractTask for gcloud cli plugin

Feature description

Could we remove the requirement for the service Account key and extend AbstractTask vs Task for authorizing for gcloud cli plugin? This way the gcloud plugin will authenticate the same way as the other plugins.

Error when the table has no data

Expected Behavior

When the query result comes to zero, the process could be improved so as not to give an error, but a success and notify that the query is coming to zero. it could be in the mysql query plugin (that's what I'm using) or in the bigquery load.

Actual Behaviour

No response

Steps To Reproduce

No response

Environment Information

Kestra Version: 0.6.1-SNAPSHOT
Plugin version: Plugin jdbc mysql query / Plugin jdbc bigquery load
Operating System (OS / Docker / Kubernetes): Docker image:kestra/kestra:develop-full
Java Version (If not docker):

Example flow

No response

Bigquery Load: empty file will lead to NullPointerException

If the from table is empty (no rows), this exception is raise :

BigQuery StorageWrite tasks gives Authentication error

Expected Behavior

BigQuery StorageWrite task should run successfully.

Actual Behaviour

BigQuery StorageWrite task gives authentication error. Here is the error stacktrace:

2024-03-23 11:27:38.790 • Using service account: smantri-kestra-20230317@<project-id>.iam.gserviceaccount.com
2024-03-23 11:27:38.794 • Using service account: smantri-kestra-20230317@<project-id>.iam.gserviceaccount.com
2024-03-23 11:27:39.739Your default credentials were not found. To set up Application Default Credentials for your environment, see https://cloud.google.com/docs/authentication/external/set-up-adc.
2024-03-23 11:27:39.739java.io.IOException: Your default credentials were not found. To set up Application Default Credentials for your environment, see https://cloud.google.com/docs/authentication/external/set-up-adc.
	at com.google.auth.oauth2.DefaultCredentialsProvider.getDefaultCredentials(DefaultCredentialsProvider.java:127)
	at com.google.auth.oauth2.GoogleCredentials.getApplicationDefault(GoogleCredentials.java:152)
	at com.google.auth.oauth2.GoogleCredentials.getApplicationDefault(GoogleCredentials.java:124)
	at com.google.api.gax.core.GoogleCredentialsProvider.getCredentials(GoogleCredentialsProvider.java:70)
	at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:162)
	at com.google.cloud.bigquery.storage.v1.stub.GrpcBigQueryWriteStub.create(GrpcBigQueryWriteStub.java:132)
	at com.google.cloud.bigquery.storage.v1.stub.BigQueryWriteStubSettings.createStub(BigQueryWriteStubSettings.java:147)
	at com.google.cloud.bigquery.storage.v1.BigQueryWriteClient.<init>(BigQueryWriteClient.java:259)
	at com.google.cloud.bigquery.storage.v1.BigQueryWriteClient.create(BigQueryWriteClient.java:241)
	at com.google.cloud.bigquery.storage.v1.ConnectionWorker.<init>(ConnectionWorker.java:330)
	at com.google.cloud.bigquery.storage.v1.StreamWriter.<init>(StreamWriter.java:235)
	at com.google.cloud.bigquery.storage.v1.StreamWriter.<init>(StreamWriter.java:57)
	at com.google.cloud.bigquery.storage.v1.StreamWriter$Builder.build(StreamWriter.java:823)
	at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter.<init>(SchemaAwareStreamWriter.java:105)
	at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter.<init>(SchemaAwareStreamWriter.java:56)
	at com.google.cloud.bigquery.storage.v1.SchemaAwareStreamWriter$Builder.build(SchemaAwareStreamWriter.java:660)
	at com.google.cloud.bigquery.storage.v1.JsonStreamWriter.<init>(JsonStreamWriter.java:50)
	at com.google.cloud.bigquery.storage.v1.JsonStreamWriter.<init>(JsonStreamWriter.java:38)
	at com.google.cloud.bigquery.storage.v1.JsonStreamWriter$Builder.build(JsonStreamWriter.java:410)
	at io.kestra.plugin.gcp.bigquery.StorageWrite.run(StorageWrite.java:128)
	at io.kestra.plugin.gcp.bigquery.StorageWrite.run(StorageWrite.java:47)
	at io.kestra.core.runners.Worker$WorkerThread.run(Worker.java:710)

Other BigQuery tasks are working as expected. Also checked that my GCP service account has access to BigQuery Storage API. Screenshot attached.

Steps To Reproduce

Took the sample flow for BigQuery StorageWrite task.
Put in the appropriate values.
Execute the flow.

Environment Information

Kestra Version: 0.15.5
Plugin version: 0.15.5
Operating System (OS / Docker / Kubernetes): Docker
Java Version (If not docker): N/A

Example flow

id: bq-storage-write
namespace: dev
tasks:
  - id: http_download
    type: io.kestra.plugin.fs.http.Download
    uri: https://raw.githubusercontent.com/kestra-io/datasets/main/csv/orders.csv
  - id: "storage_write"
    type: "io.kestra.plugin.gcp.bigquery.StorageWrite"
    from: "{{ outputs.http_download.uri }}"
    projectId: <project-id>
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    destinationTable: "<project-id>.smantri_dataset.orders_new_1"
    writeStreamType: DEFAULT

BigQuery fetch without any result error

We have a flow in error . The stack trace leads to bq plugin.
There is a missing index (empty result ?) from the current task that throws a index out of bound error.

Sample available on execution 2FU9Q1pt7EPWqlPNxLC7Aa

Bigquery Query - Can not use DECLARE when a destinationTable is provided

Expected Behavior

When I use DECLARE to implement some functions or variables for my query, I want it to work with my query that will fill my table.

Actual Behaviour

Instead, it tells me that I cannot set destination table with Scripts

Steps To Reproduce

Create a bigquery.query task
Use a DECLARE statement in the query
Execute your flow

Environment Information

Kestra Version: 0.5.2
Plugin version: 0.5.0

Example flow

id: my_id
type: com.leroymerlin.dataplatform.gcp.bigquery.Query
destinationTable: project.dataset.table
sql: |
DECLARE X int64;
set X=3;
Select X

Retry bigquery with same jobId

On retry (bigquery or kestra one), we have a job with the same id that will never work.
We need to dig if the bug is from kestra or google libs

GCS Downloads

Expected Behavior

When I move files using the xxx plugin, spaces in the file names are replaced by +, which implies that no blob is found.

Actual Behaviour

No files are moved, when file name contains spaces.

Steps To Reproduce

Run a flow with io.kestra.plugin.gcp.gcs.Downloads plugin with bucket name containing spaces.

Environment Information

Kestra Version: 0.8.0
Plugin version: 0.8.0

Example flow

id: "downloads"
type: "io.kestra.plugin.gcp.gcs.Downloads"
from: "gs://my-bucket/file names"
action: MOVE
moveDirectory: "gs://my-bucket/archives"

Add `outputFiles` output property in a structure {filename: internal_storage_uri} to the gcs.Downloads task

Feature description

same as kestra-io/plugin-aws#396

Add Eventarc trigger

Docs https://cloud.google.com/eventarc/docs

the recent addition of any HTTP endpoints inside VPC https://medium.com/google-cloud/introducing-a-new-eventarc-destination-internal-http-endpoint-in-a-vpc-network-3e347470cec6

After evaluating it, for now, Eventarc doesn't offer any promising integration options for kestra:

They don't support any custom events (i.e. event sources). The only supported event sources are GCP services (plus literally 4, not more, exactly 4 external services in the preview). There is no option to send custom events to Eventarc from kestra other than sending them via PubSub, which is already possible. Quite disappointing as I thought we could provide similar functionality to AWS Eventbridge PutEvents, but GCP doesn't support it :/
They don't support any custom destinations. The only supported one is the pre-GA HTTP endpoints inside VPC and even for that we wouldn't need to build any integration as in this case we would need to paste Kestra's webhook URL as destination HTTP endpoint URI - also, already possible.