[BEAM-2135] Move gcp-core to google-cloud-platform-core
Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/06f5a494 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/06f5a494 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/06f5a494 Branch: refs/heads/master Commit: 06f5a49400f3d4cdf809d6f939cadb47e1355d0a Parents: 9114eb3 Author: Luke Cwik <[email protected]> Authored: Tue May 2 09:11:32 2017 -0700 Committer: Luke Cwik <[email protected]> Committed: Tue May 2 10:56:32 2017 -0700 ---------------------------------------------------------------------- sdks/java/extensions/gcp-core/pom.xml | 185 ----- .../extensions/gcp/auth/CredentialFactory.java | 29 - .../gcp/auth/GcpCredentialFactory.java | 67 -- .../gcp/auth/NoopCredentialFactory.java | 68 -- .../gcp/auth/NullCredentialInitializer.java | 62 -- .../sdk/extensions/gcp/auth/package-info.java | 22 - .../options/CloudResourceManagerOptions.java | 32 - .../sdk/extensions/gcp/options/GcpOptions.java | 389 --------- .../options/GcpPipelineOptionsRegistrar.java | 39 - .../sdk/extensions/gcp/options/GcsOptions.java | 160 ---- .../gcp/options/GoogleApiDebugOptions.java | 89 --- .../extensions/gcp/options/package-info.java | 22 - .../beam/sdk/util/GcsIOChannelFactory.java | 111 --- .../sdk/util/GcsIOChannelFactoryRegistrar.java | 38 - .../apache/beam/sdk/util/GcsPathValidator.java | 98 --- .../java/org/apache/beam/sdk/util/GcsUtil.java | 796 ------------------ .../sdk/util/RetryHttpRequestInitializer.java | 192 ----- .../org/apache/beam/sdk/util/Transport.java | 122 --- .../org/apache/beam/sdk/util/gcsfs/GcsPath.java | 627 --------------- .../beam/sdk/util/gcsfs/package-info.java | 20 - .../org/apache/beam/sdk/util/package-info.java | 20 - .../extensions/gcp/GcpCoreApiSurfaceTest.java | 58 -- .../sdk/extensions/gcp/auth/TestCredential.java | 59 -- .../extensions/gcp/options/GcpOptionsTest.java | 273 ------- .../gcp/options/GoogleApiDebugOptionsTest.java | 147 ---- .../util/GcsIOChannelFactoryRegistrarTest.java | 44 - .../beam/sdk/util/GcsIOChannelFactoryTest.java | 43 - .../beam/sdk/util/GcsPathValidatorTest.java | 106 --- .../org/apache/beam/sdk/util/GcsUtilTest.java | 799 ------------------- .../util/RetryHttpRequestInitializerTest.java | 281 ------- .../apache/beam/sdk/util/gcsfs/GcsPathTest.java | 358 --------- .../google-cloud-platform-core/pom.xml | 185 +++++ .../extensions/gcp/auth/CredentialFactory.java | 29 + .../gcp/auth/GcpCredentialFactory.java | 67 ++ .../gcp/auth/NoopCredentialFactory.java | 68 ++ .../gcp/auth/NullCredentialInitializer.java | 62 ++ .../sdk/extensions/gcp/auth/package-info.java | 22 + .../options/CloudResourceManagerOptions.java | 32 + .../sdk/extensions/gcp/options/GcpOptions.java | 389 +++++++++ .../options/GcpPipelineOptionsRegistrar.java | 39 + .../sdk/extensions/gcp/options/GcsOptions.java | 160 ++++ .../gcp/options/GoogleApiDebugOptions.java | 89 +++ .../extensions/gcp/options/package-info.java | 22 + .../beam/sdk/util/GcsIOChannelFactory.java | 111 +++ .../sdk/util/GcsIOChannelFactoryRegistrar.java | 38 + .../apache/beam/sdk/util/GcsPathValidator.java | 98 +++ .../java/org/apache/beam/sdk/util/GcsUtil.java | 796 ++++++++++++++++++ .../sdk/util/RetryHttpRequestInitializer.java | 192 +++++ .../org/apache/beam/sdk/util/Transport.java | 122 +++ .../org/apache/beam/sdk/util/gcsfs/GcsPath.java | 627 +++++++++++++++ .../beam/sdk/util/gcsfs/package-info.java | 20 + .../org/apache/beam/sdk/util/package-info.java | 20 + .../extensions/gcp/GcpCoreApiSurfaceTest.java | 58 ++ .../sdk/extensions/gcp/auth/TestCredential.java | 59 ++ .../extensions/gcp/options/GcpOptionsTest.java | 273 +++++++ .../gcp/options/GoogleApiDebugOptionsTest.java | 147 ++++ .../util/GcsIOChannelFactoryRegistrarTest.java | 44 + .../beam/sdk/util/GcsIOChannelFactoryTest.java | 43 + .../beam/sdk/util/GcsPathValidatorTest.java | 106 +++ .../org/apache/beam/sdk/util/GcsUtilTest.java | 799 +++++++++++++++++++ .../util/RetryHttpRequestInitializerTest.java | 281 +++++++ .../apache/beam/sdk/util/gcsfs/GcsPathTest.java | 358 +++++++++ sdks/java/extensions/pom.xml | 2 +- 63 files changed, 5357 insertions(+), 5357 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/pom.xml ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/pom.xml b/sdks/java/extensions/gcp-core/pom.xml deleted file mode 100644 index 714cb0f..0000000 --- a/sdks/java/extensions/gcp-core/pom.xml +++ /dev/null @@ -1,185 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<!-- - Licensed to the Apache Software Foundation (ASF) under one or more - contributor license agreements. See the NOTICE file distributed with - this work for additional information regarding copyright ownership. - The ASF licenses this file to You under the Apache License, Version 2.0 - (the "License"); you may not use this file except in compliance with - the License. You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. ---> -<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> - - <modelVersion>4.0.0</modelVersion> - - <parent> - <groupId>org.apache.beam</groupId> - <artifactId>beam-sdks-java-extensions-parent</artifactId> - <version>0.7.0-SNAPSHOT</version> - <relativePath>../pom.xml</relativePath> - </parent> - - <artifactId>beam-sdks-java-extensions-gcp-core</artifactId> - <name>Apache Beam :: SDKs :: Java :: Extensions :: Google Cloud Platform Core</name> - <description>Common components used to support multiple - Google Cloud Platform specific maven modules.</description> - - <packaging>jar</packaging> - - <build> - <plugins> - <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-surefire-plugin</artifactId> - <configuration> - <excludedGroups> - org.apache.beam.sdk.testing.NeedsRunner - </excludedGroups> - <systemPropertyVariables> - <beamUseDummyRunner>true</beamUseDummyRunner> - </systemPropertyVariables> - </configuration> - </plugin> - - <!-- Coverage analysis for unit tests. --> - <plugin> - <groupId>org.jacoco</groupId> - <artifactId>jacoco-maven-plugin</artifactId> - </plugin> - </plugins> - </build> - - <dependencies> - <dependency> - <groupId>org.apache.beam</groupId> - <artifactId>beam-sdks-java-core</artifactId> - </dependency> - - <dependency> - <groupId>com.google.http-client</groupId> - <artifactId>google-http-client-jackson2</artifactId> - </dependency> - - <dependency> - <groupId>com.google.auth</groupId> - <artifactId>google-auth-library-oauth2-http</artifactId> - </dependency> - - <dependency> - <groupId>com.google.api-client</groupId> - <artifactId>google-api-client</artifactId> - </dependency> - - <dependency> - <groupId>com.google.cloud.bigdataoss</groupId> - <artifactId>gcsio</artifactId> - </dependency> - - <dependency> - <groupId>com.google.cloud.bigdataoss</groupId> - <artifactId>util</artifactId> - </dependency> - - <dependency> - <groupId>com.google.apis</groupId> - <artifactId>google-api-services-cloudresourcemanager</artifactId> - </dependency> - - <dependency> - <groupId>com.google.apis</groupId> - <artifactId>google-api-services-storage</artifactId> - </dependency> - - <dependency> - <groupId>com.google.auth</groupId> - <artifactId>google-auth-library-credentials</artifactId> - </dependency> - - <dependency> - <groupId>com.google.code.findbugs</groupId> - <artifactId>jsr305</artifactId> - </dependency> - - <dependency> - <groupId>com.google.http-client</groupId> - <artifactId>google-http-client</artifactId> - </dependency> - - <dependency> - <groupId>com.google.guava</groupId> - <artifactId>guava</artifactId> - </dependency> - - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-api</artifactId> - </dependency> - - <dependency> - <groupId>joda-time</groupId> - <artifactId>joda-time</artifactId> - </dependency> - - <dependency> - <groupId>com.fasterxml.jackson.core</groupId> - <artifactId>jackson-annotations</artifactId> - </dependency> - - <dependency> - <groupId>com.fasterxml.jackson.core</groupId> - <artifactId>jackson-databind</artifactId> - </dependency> - - <!-- build dependencies --> - <dependency> - <groupId>com.google.auto.service</groupId> - <artifactId>auto-service</artifactId> - <optional>true</optional> - </dependency> - - <dependency> - <groupId>com.google.auto.value</groupId> - <artifactId>auto-value</artifactId> - <scope>provided</scope> - </dependency> - - <!-- test dependencies --> - <dependency> - <groupId>org.apache.beam</groupId> - <artifactId>beam-sdks-java-core</artifactId> - <classifier>tests</classifier> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.hamcrest</groupId> - <artifactId>hamcrest-all</artifactId> - <scope>provided</scope> - </dependency> - - <dependency> - <groupId>org.mockito</groupId> - <artifactId>mockito-all</artifactId> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>junit</groupId> - <artifactId>junit</artifactId> - <scope>provided</scope> - </dependency> - - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-jdk14</artifactId> - <scope>test</scope> - </dependency> - </dependencies> -</project> http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/CredentialFactory.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/CredentialFactory.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/CredentialFactory.java deleted file mode 100644 index 6ab7b14..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/CredentialFactory.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.auth; - -import com.google.auth.Credentials; -import java.io.IOException; -import java.security.GeneralSecurityException; - -/** - * Construct an oauth credential to be used by the SDK and the SDK workers. - */ -public interface CredentialFactory { - Credentials getCredential() throws IOException, GeneralSecurityException; -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/GcpCredentialFactory.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/GcpCredentialFactory.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/GcpCredentialFactory.java deleted file mode 100644 index f999c63..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/GcpCredentialFactory.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.auth; - -import com.google.auth.Credentials; -import com.google.auth.oauth2.GoogleCredentials; -import java.io.IOException; -import java.util.Arrays; -import java.util.List; -import org.apache.beam.sdk.options.PipelineOptions; - -/** - * Construct an oauth credential to be used by the SDK and the SDK workers. - * Returns a GCP credential. - */ -public class GcpCredentialFactory implements CredentialFactory { - /** - * The scope cloud-platform provides access to all Cloud Platform resources. - * cloud-platform isn't sufficient yet for talking to datastore so we request - * those resources separately. - * - * <p>Note that trusted scope relationships don't apply to OAuth tokens, so for - * services we access directly (GCS) as opposed to through the backend - * (BigQuery, GCE), we need to explicitly request that scope. - */ - private static final List<String> SCOPES = Arrays.asList( - "https://www.googleapis.com/auth/cloud-platform";, - "https://www.googleapis.com/auth/devstorage.full_control";, - "https://www.googleapis.com/auth/userinfo.email";, - "https://www.googleapis.com/auth/datastore";, - "https://www.googleapis.com/auth/pubsub";); - - private static final GcpCredentialFactory INSTANCE = new GcpCredentialFactory(); - - public static GcpCredentialFactory fromOptions(PipelineOptions options) { - return INSTANCE; - } - - /** - * Returns a default GCP {@link Credentials} or null when it fails. - */ - @Override - public Credentials getCredential() { - try { - return GoogleCredentials.getApplicationDefault().createScoped(SCOPES); - } catch (IOException e) { - // Ignore the exception - // Pipelines that only access to public data should be able to run without credentials. - return null; - } - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NoopCredentialFactory.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NoopCredentialFactory.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NoopCredentialFactory.java deleted file mode 100644 index 4355a10..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NoopCredentialFactory.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.auth; - -import com.google.auth.Credentials; -import java.io.IOException; -import java.net.URI; -import java.util.List; -import java.util.Map; -import org.apache.beam.sdk.options.PipelineOptions; - -/** - * Construct an oauth credential to be used by the SDK and the SDK workers. - * Always returns a null Credential object. - */ -public class NoopCredentialFactory implements CredentialFactory { - private static final NoopCredentialFactory INSTANCE = new NoopCredentialFactory(); - private static final NoopCredentials NOOP_CREDENTIALS = new NoopCredentials(); - - public static NoopCredentialFactory fromOptions(PipelineOptions options) { - return INSTANCE; - } - - @Override - public Credentials getCredential() throws IOException { - return NOOP_CREDENTIALS; - } - - private static class NoopCredentials extends Credentials { - @Override - public String getAuthenticationType() { - return null; - } - - @Override - public Map<String, List<String>> getRequestMetadata(URI uri) throws IOException { - return null; - } - - @Override - public boolean hasRequestMetadata() { - return false; - } - - @Override - public boolean hasRequestMetadataOnly() { - return false; - } - - @Override - public void refresh() throws IOException {} - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NullCredentialInitializer.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NullCredentialInitializer.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NullCredentialInitializer.java deleted file mode 100644 index 00306f2..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/NullCredentialInitializer.java +++ /dev/null @@ -1,62 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one -* or more contributor license agreements. See the NOTICE file -* distributed with this work for additional information -* regarding copyright ownership. The ASF licenses this file -* to you under the Apache License, Version 2.0 (the -* "License"); you may not use this file except in compliance -* with the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, software -* distributed under the License is distributed on an "AS IS" BASIS, -* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -* See the License for the specific language governing permissions and -* limitations under the License. -*/ -package org.apache.beam.sdk.extensions.gcp.auth; - -import com.google.api.client.http.HttpRequest; -import com.google.api.client.http.HttpRequestInitializer; -import com.google.api.client.http.HttpResponse; -import com.google.api.client.http.HttpUnsuccessfulResponseHandler; -import java.io.IOException; - -/** - * A {@link HttpRequestInitializer} for requests that don't have credentials. - * - * <p>When the access is denied, it throws {@link IOException} with a detailed error message. - */ -public class NullCredentialInitializer implements HttpRequestInitializer { - private static final int ACCESS_DENIED = 401; - private static final String NULL_CREDENTIAL_REASON = - "Unable to get application default credentials. Please see " - + "https://developers.google.com/accounts/docs/application-default-credentials " - + "for details on how to specify credentials. This version of the SDK is " - + "dependent on the gcloud core component version 2015.02.05 or newer to " - + "be able to get credentials from the currently authorized user via gcloud auth."; - - @Override - public void initialize(HttpRequest httpRequest) throws IOException { - httpRequest.setUnsuccessfulResponseHandler(new NullCredentialHttpUnsuccessfulResponseHandler()); - } - - private static class NullCredentialHttpUnsuccessfulResponseHandler - implements HttpUnsuccessfulResponseHandler { - - @Override - public boolean handleResponse( - HttpRequest httpRequest, - HttpResponse httpResponse, boolean supportsRetry) throws IOException { - if (!httpResponse.isSuccessStatusCode() && httpResponse.getStatusCode() == ACCESS_DENIED) { - throwNullCredentialException(); - } - return supportsRetry; - } - } - - public static void throwNullCredentialException() { - throw new RuntimeException(NULL_CREDENTIAL_REASON); - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/package-info.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/package-info.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/package-info.java deleted file mode 100644 index 3d77bf2..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/auth/package-info.java +++ /dev/null @@ -1,22 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -/** - * Defines classes related to interacting with {@link com.google.auth.Credentials} for - * pipeline creation and execution containing Google Cloud Platform components. - */ -package org.apache.beam.sdk.extensions.gcp.auth; http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/CloudResourceManagerOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/CloudResourceManagerOptions.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/CloudResourceManagerOptions.java deleted file mode 100644 index 87557e5..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/CloudResourceManagerOptions.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.options; - -import org.apache.beam.sdk.options.ApplicationNameOptions; -import org.apache.beam.sdk.options.Description; -import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.options.StreamingOptions; - -/** - * Properties needed when using Google CloudResourceManager with the Apache Beam SDK. - */ -@Description("Options that are used to configure Google CloudResourceManager. See " - + "https://cloud.google.com/resource-manager/ for details on CloudResourceManager.") -public interface CloudResourceManagerOptions extends ApplicationNameOptions, GcpOptions, - PipelineOptions, StreamingOptions { -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpOptions.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpOptions.java deleted file mode 100644 index 445f00f..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpOptions.java +++ /dev/null @@ -1,389 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.options; - -import static com.google.common.base.Preconditions.checkArgument; -import static com.google.common.base.Strings.isNullOrEmpty; - -import com.fasterxml.jackson.annotation.JsonIgnore; -import com.google.api.client.http.HttpRequestInitializer; -import com.google.api.client.util.BackOff; -import com.google.api.client.util.Sleeper; -import com.google.api.services.cloudresourcemanager.CloudResourceManager; -import com.google.api.services.cloudresourcemanager.model.Project; -import com.google.api.services.storage.model.Bucket; -import com.google.auth.Credentials; -import com.google.auth.http.HttpCredentialsAdapter; -import com.google.cloud.hadoop.util.ChainingHttpRequestInitializer; -import com.google.cloud.hadoop.util.ResilientOperation; -import com.google.cloud.hadoop.util.RetryDeterminer; -import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.ImmutableList; -import com.google.common.io.Files; -import java.io.File; -import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.nio.file.FileAlreadyExistsException; -import java.security.GeneralSecurityException; -import java.util.Locale; -import java.util.Map; -import java.util.regex.Matcher; -import java.util.regex.Pattern; -import javax.annotation.Nullable; -import org.apache.beam.sdk.extensions.gcp.auth.CredentialFactory; -import org.apache.beam.sdk.extensions.gcp.auth.GcpCredentialFactory; -import org.apache.beam.sdk.extensions.gcp.auth.NullCredentialInitializer; -import org.apache.beam.sdk.options.Default; -import org.apache.beam.sdk.options.DefaultValueFactory; -import org.apache.beam.sdk.options.Description; -import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.util.FluentBackoff; -import org.apache.beam.sdk.util.InstanceBuilder; -import org.apache.beam.sdk.util.PathValidator; -import org.apache.beam.sdk.util.RetryHttpRequestInitializer; -import org.apache.beam.sdk.util.Transport; -import org.apache.beam.sdk.util.gcsfs.GcsPath; -import org.joda.time.Duration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Options used to configure Google Cloud Platform specific options such as the project - * and credentials. - * - * <p>These options defer to the - * <a href="https://developers.google.com/accounts/docs/application-default-credentials";> - * application default credentials</a> for authentication. See the - * <a href="https://github.com/google/google-auth-library-java";>Google Auth Library</a> for - * alternative mechanisms for creating credentials. - */ -@Description("Options used to configure Google Cloud Platform project and credentials.") -public interface GcpOptions extends GoogleApiDebugOptions, PipelineOptions { - /** - * Project id to use when launching jobs. - */ - @Description("Project id. Required when using Google Cloud Platform services. " - + "See https://cloud.google.com/storage/docs/projects for further details.") - @Default.InstanceFactory(DefaultProjectFactory.class) - String getProject(); - void setProject(String value); - - /** - * GCP <a href="https://developers.google.com/compute/docs/zones"; - * >availability zone</a> for operations. - * - * <p>Default is set on a per-service basis. - */ - @Description("GCP availability zone for running GCP operations. " - + "Default is up to the individual service.") - String getZone(); - void setZone(String value); - - /** - * The class of the credential factory that should be created and used to create - * credentials. If gcpCredential has not been set explicitly, an instance of this class will - * be constructed and used as a credential factory. - */ - @Description("The class of the credential factory that should be created and used to create " - + "credentials. If gcpCredential has not been set explicitly, an instance of this class will " - + "be constructed and used as a credential factory.") - @Default.Class(GcpCredentialFactory.class) - Class<? extends CredentialFactory> getCredentialFactoryClass(); - void setCredentialFactoryClass( - Class<? extends CredentialFactory> credentialFactoryClass); - - /** - * The credential instance that should be used to authenticate against GCP services. - * If no credential has been set explicitly, the default is to use the instance factory - * that constructs a credential based upon the currently set credentialFactoryClass. - */ - @JsonIgnore - @Description("The credential instance that should be used to authenticate against GCP services. " - + "If no credential has been set explicitly, the default is to use the instance factory " - + "that constructs a credential based upon the currently set credentialFactoryClass.") - @Default.InstanceFactory(GcpUserCredentialsFactory.class) - Credentials getGcpCredential(); - void setGcpCredential(Credentials value); - - /** - * Attempts to infer the default project based upon the environment this application - * is executing within. Currently this only supports getting the default project from gcloud. - */ - class DefaultProjectFactory implements DefaultValueFactory<String> { - private static final Logger LOG = LoggerFactory.getLogger(DefaultProjectFactory.class); - - @Override - public String create(PipelineOptions options) { - try { - File configFile; - if (getEnvironment().containsKey("CLOUDSDK_CONFIG")) { - configFile = new File(getEnvironment().get("CLOUDSDK_CONFIG"), "properties"); - } else if (isWindows() && getEnvironment().containsKey("APPDATA")) { - configFile = new File(getEnvironment().get("APPDATA"), "gcloud/properties"); - } else { - // New versions of gcloud use this file - configFile = new File( - System.getProperty("user.home"), - ".config/gcloud/configurations/config_default"); - if (!configFile.exists()) { - // Old versions of gcloud use this file - configFile = new File(System.getProperty("user.home"), ".config/gcloud/properties"); - } - } - String section = null; - Pattern projectPattern = Pattern.compile("^project\\s*=\\s*(.*)$"); - Pattern sectionPattern = Pattern.compile("^\\[(.*)\\]$"); - for (String line : Files.readLines(configFile, StandardCharsets.UTF_8)) { - line = line.trim(); - if (line.isEmpty() || line.startsWith(";")) { - continue; - } - Matcher matcher = sectionPattern.matcher(line); - if (matcher.matches()) { - section = matcher.group(1); - } else if (section == null || section.equals("core")) { - matcher = projectPattern.matcher(line); - if (matcher.matches()) { - String project = matcher.group(1).trim(); - LOG.info("Inferred default GCP project '{}' from gcloud. If this is the incorrect " - + "project, please cancel this Pipeline and specify the command-line " - + "argument --project.", project); - return project; - } - } - } - } catch (IOException expected) { - LOG.debug("Failed to find default project.", expected); - } - // return null if can't determine - return null; - } - - /** - * Returns true if running on the Windows OS. - */ - private static boolean isWindows() { - return System.getProperty("os.name").toLowerCase(Locale.ENGLISH).contains("windows"); - } - - /** - * Used to mock out getting environment variables. - */ - @VisibleForTesting - Map<String, String> getEnvironment() { - return System.getenv(); - } - } - - /** - * Attempts to load the GCP credentials. See - * {@link CredentialFactory#getCredential()} for more details. - */ - class GcpUserCredentialsFactory implements DefaultValueFactory<Credentials> { - @Override - public Credentials create(PipelineOptions options) { - GcpOptions gcpOptions = options.as(GcpOptions.class); - try { - CredentialFactory factory = InstanceBuilder.ofType(CredentialFactory.class) - .fromClass(gcpOptions.getCredentialFactoryClass()) - .fromFactoryMethod("fromOptions") - .withArg(PipelineOptions.class, options) - .build(); - return factory.getCredential(); - } catch (IOException | GeneralSecurityException e) { - throw new RuntimeException("Unable to obtain credential", e); - } - } - } - - /** - * A GCS path for storing temporary files in GCP. - * - * <p>Its default to {@link PipelineOptions#getTempLocation}. - */ - @Description("A GCS path for storing temporary files in GCP.") - @Default.InstanceFactory(GcpTempLocationFactory.class) - @Nullable String getGcpTempLocation(); - void setGcpTempLocation(String value); - - /** - * Returns {@link PipelineOptions#getTempLocation} as the default GCP temp location. - */ - class GcpTempLocationFactory implements DefaultValueFactory<String> { - private static final FluentBackoff BACKOFF_FACTORY = - FluentBackoff.DEFAULT.withMaxRetries(3).withInitialBackoff(Duration.millis(200)); - static final String DEFAULT_REGION = "us-central1"; - static final Logger LOG = LoggerFactory.getLogger(GcpTempLocationFactory.class); - - @Override - @Nullable - public String create(PipelineOptions options) { - String tempLocation = options.getTempLocation(); - if (isNullOrEmpty(tempLocation)) { - tempLocation = tryCreateDefaultBucket(options, - newCloudResourceManagerClient(options.as(CloudResourceManagerOptions.class)).build()); - options.setTempLocation(tempLocation); - } else { - try { - PathValidator validator = options.as(GcsOptions.class).getPathValidator(); - validator.validateOutputFilePrefixSupported(tempLocation); - } catch (Exception e) { - throw new IllegalArgumentException(String.format( - "Error constructing default value for gcpTempLocation: tempLocation is not" - + " a valid GCS path, %s. ", tempLocation), e); - } - } - return tempLocation; - } - - /** - * Creates a default bucket or verifies the existence and proper access control - * of an existing default bucket. Returns the location if successful. - */ - @VisibleForTesting - static String tryCreateDefaultBucket( - PipelineOptions options, CloudResourceManager crmClient) { - GcsOptions gcpOptions = options.as(GcsOptions.class); - - final String projectId = gcpOptions.getProject(); - checkArgument(!isNullOrEmpty(projectId), - "--project is a required option."); - - // Look up the project number, to create a default bucket with a stable - // name with no special characters. - long projectNumber = 0L; - try { - projectNumber = getProjectNumber(projectId, crmClient); - } catch (IOException e) { - throw new RuntimeException("Unable to verify project with ID " + projectId, e); - } - String region = DEFAULT_REGION; - if (!isNullOrEmpty(gcpOptions.getZone())) { - region = getRegionFromZone(gcpOptions.getZone()); - } - final String bucketName = - "dataflow-staging-" + region + "-" + projectNumber; - LOG.info("No staging location provided, attempting to use default bucket: {}", - bucketName); - Bucket bucket = new Bucket() - .setName(bucketName) - .setLocation(region); - // Always try to create the bucket before checking access, so that we do not - // race with other pipelines that may be attempting to do the same thing. - try { - gcpOptions.getGcsUtil().createBucket(projectId, bucket); - } catch (FileAlreadyExistsException e) { - LOG.debug("Bucket '{}'' already exists, verifying access.", bucketName); - } catch (IOException e) { - throw new RuntimeException("Unable create default bucket.", e); - } - - // Once the bucket is expected to exist, verify that it is correctly owned - // by the project executing the job. - try { - long owner = gcpOptions.getGcsUtil().bucketOwner( - GcsPath.fromComponents(bucketName, "")); - checkArgument( - owner == projectNumber, - "Bucket owner does not match the project from --project:" - + " %s vs. %s", owner, projectNumber); - } catch (IOException e) { - throw new RuntimeException( - "Unable to determine the owner of the default bucket at gs://" + bucketName, e); - } - return "gs://" + bucketName; - } - - /** - * Returns the project number or throws an exception if the project does not - * exist or has other access exceptions. - */ - private static long getProjectNumber( - String projectId, - CloudResourceManager crmClient) throws IOException { - return getProjectNumber( - projectId, - crmClient, - BACKOFF_FACTORY.backoff(), - Sleeper.DEFAULT); - } - - /** - * Returns the project number or throws an error if the project does not - * exist or has other access errors. - */ - private static long getProjectNumber( - String projectId, - CloudResourceManager crmClient, - BackOff backoff, - Sleeper sleeper) throws IOException { - CloudResourceManager.Projects.Get getProject = - crmClient.projects().get(projectId); - try { - Project project = ResilientOperation.retry( - ResilientOperation.getGoogleRequestCallable(getProject), - backoff, - RetryDeterminer.SOCKET_ERRORS, - IOException.class, - sleeper); - return project.getProjectNumber(); - } catch (Exception e) { - throw new IOException("Unable to get project number", e); - } - } - - @VisibleForTesting - static String getRegionFromZone(String zone) { - String[] zoneParts = zone.split("-"); - checkArgument(zoneParts.length >= 2, "Invalid zone provided: %s", zone); - return zoneParts[0] + "-" + zoneParts[1]; - } - - /** - * Returns a CloudResourceManager client builder using the specified - * {@link CloudResourceManagerOptions}. - */ - @VisibleForTesting - static CloudResourceManager.Builder newCloudResourceManagerClient( - CloudResourceManagerOptions options) { - Credentials credentials = options.getGcpCredential(); - if (credentials == null) { - NullCredentialInitializer.throwNullCredentialException(); - } - return new CloudResourceManager.Builder(Transport.getTransport(), Transport.getJsonFactory(), - chainHttpRequestInitializer( - credentials, - // Do not log 404. It clutters the output and is possibly even required by the caller. - new RetryHttpRequestInitializer(ImmutableList.of(404)))) - .setApplicationName(options.getAppName()) - .setGoogleClientRequestInitializer(options.getGoogleApiTrace()); - } - - private static HttpRequestInitializer chainHttpRequestInitializer( - Credentials credential, HttpRequestInitializer httpRequestInitializer) { - if (credential == null) { - return new ChainingHttpRequestInitializer( - new NullCredentialInitializer(), httpRequestInitializer); - } else { - return new ChainingHttpRequestInitializer( - new HttpCredentialsAdapter(credential), - httpRequestInitializer); - } - } - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpPipelineOptionsRegistrar.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpPipelineOptionsRegistrar.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpPipelineOptionsRegistrar.java deleted file mode 100644 index afc3416..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcpPipelineOptionsRegistrar.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.beam.sdk.extensions.gcp.options; - -import com.google.auto.service.AutoService; -import com.google.common.collect.ImmutableList; -import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.options.PipelineOptionsRegistrar; - -/** - * A registrar containing the default GCP options. - */ -@AutoService(PipelineOptionsRegistrar.class) -public class GcpPipelineOptionsRegistrar implements PipelineOptionsRegistrar { - @Override - public Iterable<Class<? extends PipelineOptions>> getPipelineOptions() { - return ImmutableList.<Class<? extends PipelineOptions>>builder() - .add(GcpOptions.class) - .add(GcsOptions.class) - .add(GoogleApiDebugOptions.class) - .build(); - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java deleted file mode 100644 index 954092c..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java +++ /dev/null @@ -1,160 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.options; - -import com.fasterxml.jackson.annotation.JsonIgnore; -import com.google.cloud.hadoop.util.AbstractGoogleAsyncWriteChannel; -import com.google.common.util.concurrent.MoreExecutors; -import com.google.common.util.concurrent.ThreadFactoryBuilder; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import javax.annotation.Nullable; -import org.apache.beam.sdk.options.ApplicationNameOptions; -import org.apache.beam.sdk.options.Default; -import org.apache.beam.sdk.options.DefaultValueFactory; -import org.apache.beam.sdk.options.Description; -import org.apache.beam.sdk.options.Hidden; -import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.util.GcsPathValidator; -import org.apache.beam.sdk.util.GcsUtil; -import org.apache.beam.sdk.util.InstanceBuilder; -import org.apache.beam.sdk.util.PathValidator; - -/** - * Options used to configure Google Cloud Storage. - */ -public interface GcsOptions extends - ApplicationNameOptions, GcpOptions, PipelineOptions { - /** - * The GcsUtil instance that should be used to communicate with Google Cloud Storage. - */ - @JsonIgnore - @Description("The GcsUtil instance that should be used to communicate with Google Cloud Storage.") - @Default.InstanceFactory(GcsUtil.GcsUtilFactory.class) - @Hidden - GcsUtil getGcsUtil(); - void setGcsUtil(GcsUtil value); - - /** - * The ExecutorService instance to use to create threads, can be overridden to specify an - * ExecutorService that is compatible with the users environment. If unset, the - * default is to create an ExecutorService with an unbounded number of threads; this - * is compatible with Google AppEngine. - */ - @JsonIgnore - @Description("The ExecutorService instance to use to create multiple threads. Can be overridden " - + "to specify an ExecutorService that is compatible with the users environment. If unset, " - + "the default is to create an ExecutorService with an unbounded number of threads; this " - + "is compatible with Google AppEngine.") - @Default.InstanceFactory(ExecutorServiceFactory.class) - @Hidden - ExecutorService getExecutorService(); - void setExecutorService(ExecutorService value); - - /** - * GCS endpoint to use. If unspecified, uses the default endpoint. - */ - @JsonIgnore - @Hidden - @Description("The URL for the GCS API.") - String getGcsEndpoint(); - void setGcsEndpoint(String value); - - /** - * The buffer size (in bytes) to use when uploading files to GCS. Please see the documentation for - * {@link AbstractGoogleAsyncWriteChannel#setUploadBufferSize} for more information on the - * restrictions and performance implications of this value. - */ - @Description("The buffer size (in bytes) to use when uploading files to GCS. Please see the " - + "documentation for AbstractGoogleAsyncWriteChannel.setUploadBufferSize for more " - + "information on the restrictions and performance implications of this value.\n\n" - + "https://github.com/GoogleCloudPlatform/bigdata-interop/blob/master/util/src/main/java/"; - + "com/google/cloud/hadoop/util/AbstractGoogleAsyncWriteChannel.java") - @Nullable - Integer getGcsUploadBufferSizeBytes(); - void setGcsUploadBufferSizeBytes(@Nullable Integer bytes); - - /** - * The class of the validator that should be created and used to validate paths. - * If pathValidator has not been set explicitly, an instance of this class will be - * constructed and used as the path validator. - */ - @Description("The class of the validator that should be created and used to validate paths. " - + "If pathValidator has not been set explicitly, an instance of this class will be " - + "constructed and used as the path validator.") - @Default.Class(GcsPathValidator.class) - Class<? extends PathValidator> getPathValidatorClass(); - void setPathValidatorClass(Class<? extends PathValidator> validatorClass); - - /** - * The path validator instance that should be used to validate paths. - * If no path validator has been set explicitly, the default is to use the instance factory that - * constructs a path validator based upon the currently set pathValidatorClass. - */ - @JsonIgnore - @Description("The path validator instance that should be used to validate paths. " - + "If no path validator has been set explicitly, the default is to use the instance factory " - + "that constructs a path validator based upon the currently set pathValidatorClass.") - @Default.InstanceFactory(PathValidatorFactory.class) - PathValidator getPathValidator(); - void setPathValidator(PathValidator validator); - - /** - * Returns the default {@link ExecutorService} to use within the Apache Beam SDK. The - * {@link ExecutorService} is compatible with AppEngine. - */ - class ExecutorServiceFactory implements DefaultValueFactory<ExecutorService> { - @SuppressWarnings("deprecation") // IS_APP_ENGINE is deprecated for internal use only. - @Override - public ExecutorService create(PipelineOptions options) { - ThreadFactoryBuilder threadFactoryBuilder = new ThreadFactoryBuilder(); - threadFactoryBuilder.setThreadFactory(MoreExecutors.platformThreadFactory()); - threadFactoryBuilder.setDaemon(true); - /* The SDK requires an unbounded thread pool because a step may create X writers - * each requiring their own thread to perform the writes otherwise a writer may - * block causing deadlock for the step because the writers buffer is full. - * Also, the MapTaskExecutor launches the steps in reverse order and completes - * them in forward order thus requiring enough threads so that each step's writers - * can be active. - */ - return new ThreadPoolExecutor( - 0, Integer.MAX_VALUE, // Allow an unlimited number of re-usable threads. - Long.MAX_VALUE, TimeUnit.NANOSECONDS, // Keep non-core threads alive forever. - new SynchronousQueue<Runnable>(), - threadFactoryBuilder.build()); - } - } - - /** - * Creates a {@link PathValidator} object using the class specified in - * {@link #getPathValidatorClass()}. - */ - class PathValidatorFactory implements DefaultValueFactory<PathValidator> { - @Override - public PathValidator create(PipelineOptions options) { - GcsOptions gcsOptions = options.as(GcsOptions.class); - return InstanceBuilder.ofType(PathValidator.class) - .fromClass(gcsOptions.getPathValidatorClass()) - .fromFactoryMethod("fromOptions") - .withArg(PipelineOptions.class, options) - .build(); - } - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GoogleApiDebugOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GoogleApiDebugOptions.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GoogleApiDebugOptions.java deleted file mode 100644 index 01144c4..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GoogleApiDebugOptions.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.extensions.gcp.options; - -import com.google.api.client.googleapis.services.AbstractGoogleClient; -import com.google.api.client.googleapis.services.AbstractGoogleClientRequest; -import com.google.api.client.googleapis.services.GoogleClientRequestInitializer; -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; -import org.apache.beam.sdk.options.Description; -import org.apache.beam.sdk.options.PipelineOptions; - -/** - * These options configure debug settings for Google API clients created within the Apache Beam SDK. - */ -public interface GoogleApiDebugOptions extends PipelineOptions { - /** - * This option enables tracing of API calls to Google services used within the Apache - * Beam SDK. Values are expected in JSON format <code>{"ApiName":"TraceDestination",...} - * </code> where the {@code ApiName} represents the request classes canonical name. The - * {@code TraceDestination} is a logical trace consumer to whom the trace will be reported. - * Typically, "producer" is the right destination to use: this makes API traces available to the - * team offering the API. Note that by enabling this option, the contents of the requests to and - * from Google Cloud services will be made available to Google. For example, by specifying - * <code>{"Dataflow":"producer"}</code>, all calls to the Dataflow service will be made available - * to Google, specifically to the Google Cloud Dataflow team. - */ - @Description("This option enables tracing of API calls to Google services used within the Apache " - + "Beam SDK. Values are expected in JSON format {\"ApiName\":\"TraceDestination\",...} " - + "where the ApiName represents the request classes canonical name. The TraceDestination is " - + "a logical trace consumer to whom the trace will be reported. Typically, \"producer\" is " - + "the right destination to use: this makes API traces available to the team offering the " - + "API. Note that by enabling this option, the contents of the requests to and from " - + "Google Cloud services will be made available to Google. For example, by specifying " - + "{\"Dataflow\":\"producer\"}, all calls to the Dataflow service will be made available to " - + "Google, specifically to the Google Cloud Dataflow team.") - GoogleApiTracer getGoogleApiTrace(); - void setGoogleApiTrace(GoogleApiTracer commands); - - /** - * A {@link GoogleClientRequestInitializer} that adds the trace destination to Google API calls. - */ - class GoogleApiTracer extends HashMap<String, String> - implements GoogleClientRequestInitializer { - /** - * Creates a {@link GoogleApiTracer} that sets the trace destination on all - * calls that match the given client type. - */ - public GoogleApiTracer addTraceFor(AbstractGoogleClient client, String traceDestination) { - put(client.getClass().getCanonicalName(), traceDestination); - return this; - } - - /** - * Creates a {@link GoogleApiTracer} that sets the trace {@code traceDestination} on all - * calls that match for the given request type. - */ - public GoogleApiTracer addTraceFor( - AbstractGoogleClientRequest<?> request, String traceDestination) { - put(request.getClass().getCanonicalName(), traceDestination); - return this; - } - - @Override - public void initialize(AbstractGoogleClientRequest<?> request) throws IOException { - for (Map.Entry<String, String> entry : this.entrySet()) { - if (request.getClass().getCanonicalName().contains(entry.getKey())) { - request.set("$trace", entry.getValue()); - } - } - } - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/package-info.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/package-info.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/package-info.java deleted file mode 100644 index bc9646c..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/package-info.java +++ /dev/null @@ -1,22 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -/** - * Defines {@link org.apache.beam.sdk.options.PipelineOptions} for - * configuring pipeline execution for Google Cloud Platform components. - */ -package org.apache.beam.sdk.extensions.gcp.options; http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java deleted file mode 100644 index 3a12620..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java +++ /dev/null @@ -1,111 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import java.io.IOException; -import java.nio.channels.ReadableByteChannel; -import java.nio.channels.WritableByteChannel; -import java.nio.file.Path; -import java.util.Collection; -import java.util.LinkedList; -import java.util.List; -import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; -import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.util.gcsfs.GcsPath; - -/** - * Implements IOChannelFactory for GCS. - */ -public class GcsIOChannelFactory implements IOChannelFactory { - - /** - * Create a {@link GcsIOChannelFactory} with the given {@link PipelineOptions}. - */ - public static GcsIOChannelFactory fromOptions(PipelineOptions options) { - return new GcsIOChannelFactory(options.as(GcsOptions.class)); - } - - private final GcsOptions options; - - private GcsIOChannelFactory(GcsOptions options) { - this.options = options; - } - - @Override - public Collection<String> match(String spec) throws IOException { - GcsPath path = GcsPath.fromUri(spec); - GcsUtil util = options.getGcsUtil(); - List<GcsPath> matched = util.expand(path); - - List<String> specs = new LinkedList<>(); - for (GcsPath match : matched) { - specs.add(match.toString()); - } - - return specs; - } - - @Override - public ReadableByteChannel open(String spec) throws IOException { - GcsPath path = GcsPath.fromUri(spec); - GcsUtil util = options.getGcsUtil(); - return util.open(path); - } - - @Override - public WritableByteChannel create(String spec, String mimeType) - throws IOException { - GcsPath path = GcsPath.fromUri(spec); - GcsUtil util = options.getGcsUtil(); - return util.create(path, mimeType); - } - - @Override - public long getSizeBytes(String spec) throws IOException { - GcsPath path = GcsPath.fromUri(spec); - GcsUtil util = options.getGcsUtil(); - return util.fileSize(path); - } - - @Override - public boolean isReadSeekEfficient(String spec) throws IOException { - // TODO It is incorrect to return true here for files with content encoding set to gzip. - return true; - } - - @Override - public String resolve(String path, String other) throws IOException { - return toPath(path).resolve(other).toString(); - } - - @Override - public Path toPath(String path) { - return GcsPath.fromUri(path); - } - - @Override - public void copy(Iterable<String> srcFilenames, Iterable<String> destFilenames) - throws IOException { - options.getGcsUtil().copy(srcFilenames, destFilenames); - } - - @Override - public void remove(Collection<String> filesOrDirs) throws IOException { - options.getGcsUtil().remove(filesOrDirs); - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java deleted file mode 100644 index b4c457f..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import com.google.auto.service.AutoService; -import org.apache.beam.sdk.options.PipelineOptions; - -/** - * {@link AutoService} registrar for the {@link GcsIOChannelFactory}. - */ -@AutoService(IOChannelFactoryRegistrar.class) -public class GcsIOChannelFactoryRegistrar implements IOChannelFactoryRegistrar { - - @Override - public GcsIOChannelFactory fromOptions(PipelineOptions options) { - return GcsIOChannelFactory.fromOptions(options); - } - - @Override - public String getScheme() { - return "gs"; - } -} http://git-wip-us.apache.org/repos/asf/beam/blob/06f5a494/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java ---------------------------------------------------------------------- diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java deleted file mode 100644 index 4d58424..0000000 --- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java +++ /dev/null @@ -1,98 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.beam.sdk.util; - -import static com.google.common.base.Preconditions.checkArgument; - -import java.io.IOException; -import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; -import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.util.gcsfs.GcsPath; - -/** - * GCP implementation of {@link PathValidator}. Only GCS paths are allowed. - */ -public class GcsPathValidator implements PathValidator { - - private GcsOptions gcpOptions; - - private GcsPathValidator(GcsOptions options) { - this.gcpOptions = options; - } - - public static GcsPathValidator fromOptions(PipelineOptions options) { - return new GcsPathValidator(options.as(GcsOptions.class)); - } - - /** - * Validates the the input GCS path is accessible and that the path - * is well formed. - */ - @Override - public String validateInputFilePatternSupported(String filepattern) { - GcsPath gcsPath = getGcsPath(filepattern); - checkArgument(gcpOptions.getGcsUtil().isGcsPatternSupported(gcsPath.getObject())); - String returnValue = verifyPath(filepattern); - verifyPathIsAccessible(filepattern, "Could not find file %s"); - return returnValue; - } - - /** - * Validates the the output GCS path is accessible and that the path - * is well formed. - */ - @Override - public String validateOutputFilePrefixSupported(String filePrefix) { - String returnValue = verifyPath(filePrefix); - verifyPathIsAccessible(filePrefix, "Output path does not exist or is not writeable: %s"); - return returnValue; - } - - @Override - public String verifyPath(String path) { - GcsPath gcsPath = getGcsPath(path); - checkArgument(gcsPath.isAbsolute(), "Must provide absolute paths for Dataflow"); - checkArgument(!gcsPath.getObject().isEmpty(), - "Missing object or bucket in path: '%s', did you mean: 'gs://some-bucket/%s'?", - gcsPath, gcsPath.getBucket()); - checkArgument(!gcsPath.getObject().contains("//"), - "Dataflow Service does not allow objects with consecutive slashes"); - return gcsPath.toResourceName(); - } - - private void verifyPathIsAccessible(String path, String errorMessage) { - GcsPath gcsPath = getGcsPath(path); - try { - checkArgument(gcpOptions.getGcsUtil().bucketAccessible(gcsPath), - errorMessage, path); - } catch (IOException e) { - throw new RuntimeException( - String.format("Unable to verify that GCS bucket gs://%s exists.", gcsPath.getBucket()), - e); - } - } - - private GcsPath getGcsPath(String path) { - try { - return GcsPath.fromUri(path); - } catch (IllegalArgumentException e) { - throw new IllegalArgumentException(String.format( - "Expected a valid 'gs://' path but was given '%s'", path), e); - } - } -}
