일반적인 문제해결

Pub/Sub 사용 중에 문제가 발생할 경우 유용하게 활용할 수 있는 문제 해결 단계를 알아봅니다.

주제를 만들 수 없음

필요한 권한이 있는지 확인합니다. Pub/Sub 주제를 만들려면 프로젝트에 대해 Pub/Sub 편집자 (roles/pubsub.editor) Identity and Access Management 역할이 필요합니다. 이 역할이 없으면 관리자에게 문의하세요. 주제 관련 문제 해결에 대한 자세한 내용은 다음 페이지를 참고하세요.

• 문제 해결 주제

• 가져오기 주제 문제 해결

• 권한 문제 해결

구독 생성 불가

다음을 완료했는지 확인합니다.

• 필요한 권한이 있는지 확인합니다. Pub/Sub 구독을 만들려면 프로젝트에 대해 Pub/Sub 편집자(roles/pubsub.editor) IAM 역할이 필요합니다. 이 역할이 없으면 관리자에게 문의하세요.

• 구독 이름을 지정했습니다.

• 구독을 연결하려는 기존 주제 이름을 지정했습니다.

• 푸시 구독을 만드는 경우 pushEndpoint 필드에 수신 URL의 프로토콜로 https://를 소문자 (http:// 또는 HTTPS:// 아님)로 지정했습니다.

구독에 관한 문제 해결 정보는 다음 페이지를 참고하세요.

• 풀, 푸시, BigQuery 또는 Cloud Storage 문제 해결

• 단일 메시지 변환을 사용한 정기 결제 문제 해결

• 권한 문제 해결

권한 문제 해결

Pub/Sub 권한은 Pub/Sub 리소스에 대해 작업을 실행할 수 있는 사용자 및 서비스 계정을 제어합니다. 권한이 잘못 구성되면 권한 거부 오류가 발생하고 메시지 흐름이 중단될 수 있습니다. 감사 로그는 모든 권한 변경사항에 대한 자세한 기록을 제공하므로 이러한 문제의 소스를 파악할 수 있습니다.

감사 로그와 관련된 Pub/Sub 권한 문제를 해결하려면 다음 단계를 따르세요.

1. 로그 탐색기를 볼 수 있는 필수 권한을 획득합니다.

   자세한 내용은 시작하기 전에를 참조하세요.

2. Google Cloud 콘솔에서 로그 탐색기 페이지로 이동합니다.

   로그 탐색기로 이동

3. 기존 Google Cloud 프로젝트, 폴더 또는 조직을 선택합니다.

4. 관련 로그를 찾는 데 사용할 수 있는 필터 목록은 다음과 같습니다.

   • resource.type="pubsub_topic" OR resource.type="pubsub_subscription": 주제 또는 구독 구성이나 액세스 제어의 변경과 관련된 문제를 해결할 때 이 쿼리를 시작점으로 사용하세요. 다른 필터와 결합하여 검색을 더욱 세부적으로 조정할 수 있습니다.

   • protoPayload.methodName="google.iam.v1.SetIamPolicy": 문제가 잘못되었거나 누락된 권한으로 인해 발생한다고 의심되는 경우 이 쿼리를 사용하세요. IAM 정책을 변경한 사용자와 변경사항을 추적하는 데 도움이 됩니다. 이는 사용자가 주제에 게시하거나 구독을 구독할 수 없는 문제, 애플리케이션의 Pub/Sub 리소스 액세스가 거부된 문제, 액세스 제어의 예기치 않은 변경사항과 같은 문제를 해결하는 데 유용합니다.

   • protoPayload.status.code=7: 권한과 명시적으로 관련된 오류가 발생할 때 이 질문을 사용하세요. 이렇게 하면 실패한 작업과 해당 작업을 시도한 사용자를 정확하게 파악할 수 있습니다. 이 쿼리를 이전 쿼리와 결합하여 권한 거부를 유발할 수 있는 특정 리소스 및 IAM 정책 변경사항을 식별할 수 있습니다.

5. 로그를 분석하여 이벤트의 타임스탬프, 변경을 수행한 주 구성원, 적용된 변경사항의 유형과 같은 요소를 확인합니다.

6. 감사 로그에서 수집한 정보를 바탕으로 수정 조치를 취할 수 있습니다.

Terraform 권한 문제 해결

Terraform과 함께 Pub/Sub를 사용하는 경우 Terraform 코드에서 필요한 역할을 명시적으로 부여하세요. 예를 들어 게시의 경우 애플리케이션의 서비스 계정에 roles/pubsub.publisher 역할이 필요합니다. 이 역할이 Terraform 코드에 명시적으로 정의되어 있지 않으면 향후 terraform apply에서 삭제할 수 있습니다. 이는 관련 없는 업데이트 중에 자주 발생하며, 신뢰할 수 있는 애플리케이션이 갑자기 PERMISSION_DENIED 오류로 실패하게 됩니다. 코드에서 역할을 명시적으로 정의하면 이러한 실수로 인한 회귀를 방지할 수 있습니다.

구독이 삭제됨

Pub/Sub 구독은 다음 두 가지 기본 방법으로 삭제할 수 있습니다.

• 충분한 권한이 있는 사용자 또는 서비스 계정이 의도적으로 구독을 삭제합니다.

• 구독은 일정 기간(기본적으로 31일) 동안 활동이 없으면 자동으로 삭제됩니다. 정기 결제 만료 정책에 대한 자세한 내용은 만료 기간을 참고하세요.

삭제된 구독 문제를 해결하려면 다음 단계를 따르세요.

1. Google Cloud 콘솔에서 Pub/Sub 구독 페이지로 이동하여 구독이 더 이상 나열되지 않는지 확인합니다. 정기 결제를 나열하는 방법에 대한 자세한 내용은 정기 결제 나열을 참고하세요.

2. 감사 로그를 확인합니다. 로그 탐색기로 이동합니다. 필터 protoPayload.methodName="google.pubsub.v1.Subscriber.DeleteSubscription"를 사용하여 삭제된 구독을 찾습니다. 로그를 검토하여 누군가 구독을 삭제했는지 아니면 활동이 없어 삭제되었는지 확인합니다. InternalExpireInactiveSubscription는 비활성 상태로 인해 구독이 삭제되었음을 나타냅니다. 감사 로그를 사용하여 문제를 해결하는 방법에 대한 자세한 내용은 감사 로그를 사용하여 Pub/Sub 문제 해결을 참고하세요.

오류 403 (Forbidden)개

403 오류는 일반적으로 작업을 수행할 올바른 권한이 없음을 의미합니다. 예를 들어 주제에 게시하거나 구독에서 가져오려고 하면 403 User not authorized 오류가 발생할 수 있습니다.

이 오류가 발생한다면 다음을 수행하세요.

• Google Cloud 콘솔에서 Pub/Sub API를 사용 설정했는지 확인합니다.

• 특히 프로젝트 간 통신에 Pub/Sub API를 사용하는 경우, 요청을 수행하는 주 구성원이 관련 Pub/Sub API 리소스에 필요한 권한을 갖고 있는지 확인합니다.

• Dataflow를 사용하는 경우 {PROJECT_NUMBER}@cloudservices.gserviceaccount.com 및 Compute Engine 서비스 계정 {PROJECT_NUMBER}[e-mail]에 모두 관련 Pub/Sub API 리소스에 대한 필수 권한이 있는지 확인합니다. 자세한 내용은 Dataflow 보안 및 권한을 참조하세요.

• App Engine을 사용한다면 프로젝트의 권한 페이지를 확인해 App Engine 서비스 계정이 Pub/Sub 편집자로 표시되어 있는지 살펴봅니다. 그렇지 않다면 App Engine 서비스 계정을 Pub/Sub 편집자로 추가해야 합니다. 일반적으로 App Engine 서비스 계정은 <project-id>@appspot.gserviceaccount.com 형식입니다.

• 감사 로그를 사용하여 권한 문제를 해결할 수 있습니다.

기타 일반적인 오류 코드

Pub/Sub API와 관련된 기타 일반적인 오류 코드 목록과 설명은 오류 코드를 참고하세요.

과도한 관리 작업 사용

관리 작업에 할당량을 너무 많이 사용하고 있는 경우 코드를 리팩터링해야 할 수도 있습니다. 다음 유사 코드를 예로 들어 보겠습니다. 이 예시에서는 리소스를 소모하기 전에 관리 작업 (GET)을 사용하여 구독이 있는지 확인합니다. GET 및 CREATE 모두 관리 작업입니다.

if !GetSubscription my-sub {
  CreateSubscription my-sub
}
Consume from subscription my-sub

보다 효율적인 패턴은 구독에서 메시지를 소비하려고 시도하는 것입니다(구독 이름을 합리적으로 확신할 수 있다고 가정). 이 낙관적 접근 방식에서는 오류가 있는 경우에만 구독을 가져오거나 생성합니다. 다음 예를 참조하세요.

try {
  Consume from subscription my-sub
} catch NotFoundError {
  CreateSubscription my-sub
  Consume from subscription my-sub
}

다음 코드 샘플을 사용하여 원하는 언어로 이 패턴을 구현할 수 있습니다.

Go

다음 샘플에서는 Go Pub/Sub 클라이언트 라이브러리의 주요 버전 (v2)을 사용합니다. 아직 v1 라이브러리를 사용하고 있다면 v2로의 마이그레이션 가이드를 참고하세요. v1 코드 샘플 목록을 보려면 지원 중단된 코드 샘플을 참고하세요.

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 Go 설정 안내를 따르세요. 자세한 내용은 Pub/Sub Go API 참고 문서를 참조하세요.

import (
	"context"
	"errors"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// optimisticSubscribe shows the recommended pattern for optimistically
// assuming a subscription exists prior to receiving messages.
func optimisticSubscribe(w io.Writer, projectID, topic, subscriptionName string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
	// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
	// If a subscription ID is provided, the project ID from the client is used.
	sub := client.Subscriber(subscriptionName)

	// Receive messages for 10 seconds, which simplifies testing.
	// Comment this out in production, since `Receive` should
	// be used as a long running operation.
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	// Instead of checking if the subscription exists, optimistically try to
	// receive from the subscription assuming it exists.
	err = sub.Receive(ctx, func(_ context.Context, msg *pubsub.Message) {
		fmt.Fprintf(w, "Got from existing subscription: %q\n", string(msg.Data))
		msg.Ack()
	})
	if err != nil {
		if st, ok := status.FromError(err); ok {
			if st.Code() == codes.NotFound {
				// If the subscription does not exist, then create the subscription.
				subscription, err := client.SubscriptionAdminClient.CreateSubscription(ctx, &pubsubpb.Subscription{
					Name:  subscriptionName,
					Topic: topic,
				})
				if err != nil {
					return err
				}
				fmt.Fprintf(w, "Created subscription: %q\n", subscriptionName)

				// client.Subscriber can be passed a subscription ID (e.g. "my-sub") or
				// a fully qualified name (e.g. "projects/my-project/subscriptions/my-sub").
				// If a subscription ID is provided, the project ID from the client is used.
				sub = client.Subscriber(subscription.GetName())
				err = sub.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
					fmt.Fprintf(w, "Got from new subscription: %q\n", string(msg.Data))
					msg.Ack()
				})
				if err != nil && !errors.Is(err, context.Canceled) {
					return err
				}
			}
		}
	}
	return nil
}

자바

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 Java 설정 안내를 따르세요. 자세한 내용은 Pub/Sub 자바 API 참조 문서를 참조하세요.

import com.google.api.gax.rpc.NotFoundException;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.PushConfig;
import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class OptimisticSubscribeExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String subscriptionId = "your-subscription-id";
    String topicId = "your-topic-id";

    optimisticSubscribeExample(projectId, subscriptionId, topicId);
  }

  public static void optimisticSubscribeExample(
      String projectId, String subscriptionId, String topicId) throws IOException {
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          // Handle incoming message, then ack the received message.
          System.out.println("Id: " + message.getMessageId());
          System.out.println("Data: " + message.getData().toStringUtf8());
          consumer.ack();
        };

    Subscriber subscriber = null;
    try {
      subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();

      // Listen for resource NOT_FOUND errors and rebuild the  subscriber and restart subscribing
      // when the current subscriber encounters these errors.
      subscriber.addListener(
          new Subscriber.Listener() {
            public void failed(Subscriber.State from, Throwable failure) {
              System.out.println(failure.getStackTrace());
              if (failure instanceof NotFoundException) {
                try (SubscriptionAdminClient subscriptionAdminClient =
                    SubscriptionAdminClient.create()) {
                  TopicName topicName = TopicName.of(projectId, topicId);
                  // Create a pull subscription with default acknowledgement deadline of 10 seconds.
                  // The client library will automatically extend acknowledgement deadlines.
                  Subscription subscription =
                      subscriptionAdminClient.createSubscription(
                          subscriptionName, topicName, PushConfig.getDefaultInstance(), 10);
                  System.out.println("Created pull subscription: " + subscription.getName());
                  optimisticSubscribeExample(projectId, subscriptionId, topicId);
                } catch (IOException err) {
                  System.out.println("Failed to create pull subscription: " + err.getMessage());
                }
              }
            }
          },
          MoreExecutors.directExecutor());

      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (IllegalStateException e) {
      // Prevent an exception from being thrown if it is the expected NotFoundException
      if (!(subscriber.failureCause() instanceof NotFoundException)) {
        throw e;
      }
    } catch (TimeoutException e) {
      subscriber.stopAsync();
    }
  }
}

Node.js

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 Node.js 설정 안내를 따르세요. 자세한 내용은 Pub/Sub Node.js API 참조 문서를 참조하세요.

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const timeout = 60;

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

function optimisticSubscribe(subscriptionNameOrId, topicNameOrId, timeout) {
  // Try using an existing subscription
  let subscription = pubSubClient.subscription(subscriptionNameOrId);

  // Create an event handler to handle messages
  let messageCount = 0;
  const messageHandler = message => {
    console.log(`Received message ${message.id}:`);
    console.log(`\tData: ${message.data}`);
    console.log(`\tAttributes: ${message.attributes}`);
    messageCount += 1;

    // "Ack" (acknowledge receipt of) the message
    message.ack();
  };

  // Set an error handler so that we're notified if the subscription doesn't
  // already exist.
  subscription.on('error', async e => {
    // Resource Not Found
    if (e.code === 5) {
      console.log('Subscription not found, creating it');
      await pubSubClient.createSubscription(
        topicNameOrId,
        subscriptionNameOrId,
      );

      // Refresh our subscriber object and re-attach the message handler.
      subscription = pubSubClient.subscription(subscriptionNameOrId);
      subscription.on('message', messageHandler);
    }
  });

  // Listen for new messages until timeout is hit; this will attempt to
  // open the actual subscriber streams. If it fails, the error handler
  // above will be called.
  subscription.on('message', messageHandler);

  // Wait a while for the subscription to run. (Part of the sample only.)
  setTimeout(() => {
    subscription.removeListener('message', messageHandler);
    console.log(`${messageCount} message(s) received.`);
  }, timeout * 1000);
}

Node.ts

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 Node.js 설정 안내를 따르세요. 자세한 내용은 Pub/Sub Node.js API 참조 문서를 참조하세요.

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const timeout = 60;

// Imports the Google Cloud client library
import {PubSub, Message, StatusError} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

function optimisticSubscribe(
  subscriptionNameOrId: string,
  topicNameOrId: string,
  timeout: number,
) {
  // Try using an existing subscription
  let subscription = pubSubClient.subscription(subscriptionNameOrId);

  // Create an event handler to handle messages
  let messageCount = 0;
  const messageHandler = (message: Message) => {
    console.log(`Received message ${message.id}:`);
    console.log(`\tData: ${message.data}`);
    console.log(`\tAttributes: ${message.attributes}`);
    messageCount += 1;

    // "Ack" (acknowledge receipt of) the message
    message.ack();
  };

  // Set an error handler so that we're notified if the subscription doesn't
  // already exist.
  subscription.on('error', async (e: StatusError) => {
    // Resource Not Found
    if (e.code === 5) {
      console.log('Subscription not found, creating it');
      await pubSubClient.createSubscription(
        topicNameOrId,
        subscriptionNameOrId,
      );

      // Refresh our subscriber object and re-attach the message handler.
      subscription = pubSubClient.subscription(subscriptionNameOrId);
      subscription.on('message', messageHandler);
    }
  });

  // Listen for new messages until timeout is hit; this will attempt to
  // open the actual subscriber streams. If it fails, the error handler
  // above will be called.
  subscription.on('message', messageHandler);

  // Wait a while for the subscription to run. (Part of the sample only.)
  setTimeout(() => {
    subscription.removeListener('message', messageHandler);
    console.log(`${messageCount} message(s) received.`);
  }, timeout * 1000);
}

Python

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 Python 설정 안내를 따르세요. 자세한 내용은 Pub/Sub Python API 참조 문서를 참조하세요.

from google.api_core.exceptions import NotFound
from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError

# TODO(developer)
# project_id = "your-project-id"
# subscription_id = "your-subscription-id"
# Number of seconds the subscriber should listen for messages
# timeout = 5.0
# topic_id = "your-topic-id"

# Create a subscriber client.
subscriber = pubsub_v1.SubscriberClient()

# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)

# Define callback to be called when a message is received.
def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    # Ack message after processing it.
    message.ack()

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # Optimistically subscribe to messages on the subscription.
        streaming_pull_future = subscriber.subscribe(
            subscription_path, callback=callback
        )
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        print("Successfully subscribed until the timeout passed.")
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.
    except NotFound:
        print(f"Subscription {subscription_path} not found, creating it.")

        try:
            # If the subscription does not exist, then create it.
            publisher = pubsub_v1.PublisherClient()
            topic_path = publisher.topic_path(project_id, topic_id)
            subscription = subscriber.create_subscription(
                request={"name": subscription_path, "topic": topic_path}
            )

            if subscription:
                print(f"Subscription {subscription.name} created")
            else:
                raise ValueError("Subscription creation failed.")

            # Subscribe on the created subscription.
            try:
                streaming_pull_future = subscriber.subscribe(
                    subscription.name, callback=callback
                )
                streaming_pull_future.result(timeout=timeout)
            except TimeoutError:
                streaming_pull_future.cancel()  # Trigger the shutdown.
                streaming_pull_future.result()  # Block until the shutdown is complete.
        except Exception as e:
            print(
                f"Exception occurred when creating subscription and subscribing to it: {e}"
            )
    except Exception as e:
        print(f"Exception occurred when attempting optimistic subscribe: {e}")

C++

이 샘플을 시도하기 전에 빠른 시작: 클라이언트 라이브러리 사용의 C++ 설정 안내를 따르세요. 자세한 내용은 Pub/Sub C++ API 참고 문서를 확인하세요.

auto process_response = [](gc::StatusOr<pubsub::PullResponse> response) {
  if (response) {
    std::cout << "Received message " << response->message << "\n";
    std::move(response->handler).ack();
    return gc::Status();
  }
  if (response.status().code() == gc::StatusCode::kUnavailable &&
      response.status().message() == "no messages returned") {
    std::cout << "No messages returned from Pull()\n";
    return gc::Status();
  }
  return response.status();
};

// Instead of checking if the subscription exists, optimistically try to
// consume from the subscription.
auto status = process_response(subscriber.Pull());
if (status.ok()) return;
if (status.code() != gc::StatusCode::kNotFound) throw std::move(status);

// Since the subscription does not exist, create the subscription.
pubsub_admin::SubscriptionAdminClient subscription_admin_client(
    pubsub_admin::MakeSubscriptionAdminConnection());
google::pubsub::v1::Subscription request;
request.set_name(
    pubsub::Subscription(project_id, subscription_id).FullName());
request.set_topic(
    pubsub::Topic(project_id, std::move(topic_id)).FullName());
auto sub = subscription_admin_client.CreateSubscription(request);
if (!sub) throw std::move(sub).status();

// Consume from the new subscription.
status = process_response(subscriber.Pull());
if (!status.ok()) throw std::move(status);