マイクロサービス入門 — サービスを分割する

$ find app -name "*.rb" | wc -l
847  # Rubyファイル847個
 
$ cloc app/
Language   Files  Blank  Comment  Code
Ruby         847   4821     1203  42891  ← 4万行超

# NG: 複数サービスが同じテーブルを参照する
# task_service/app/models/user.rb
class User < ApplicationRecord
  # task_service が auth_service の users テーブルを直接参照している！
  establish_connection :shared_production_db
end
 
# 何が問題か:
# - auth_service がスキーマを変更すると task_service が壊れる
# - 2つのサービスが同じDBに依存するので独立デプロイできない
# - これは「分散モノリス」の典型的な症状

# NG: A → B → C → D と同期的に呼び出す
# analytics_service/app/controllers/reports_controller.rb
def create_report
  # 各サービスが順番に呼び出される
  user = AuthService.get_user(params[:user_id])          # 50ms
  tasks = TaskService.get_tasks(params[:project_id])     # 80ms
  billing = BillingService.get_plan(user[:plan_id])      # 60ms
 
  # 合計190ms + 自分の処理 = レイテンシが積み重なる
  # しかも BillingService が落ちたら全体が失敗する
  generate_report(user, tasks, billing)
end

# config/routes.rb（モノリス）
# 全ルートをそのまま通過させ、まずプロキシ層を挿入
Rails.application.routes.draw do
  # 通知関連のエンドポイントだけプロキシに向ける準備をする
  constraints(NotificationProxy.new) do
    match '/api/v1/notifications/*path',
          to: 'proxy#forward',
          via: :all
  end
 
  # 他はすべてモノリスで処理
  resources :tasks
  resources :projects
end
 
# app/controllers/proxy_controller.rb
class ProxyController < ApplicationController
  NOTIFICATION_SERVICE_URL = ENV.fetch('NOTIFICATION_SERVICE_URL', nil)
 
  def forward
    if NOTIFICATION_SERVICE_URL
      forward_to_new_service
    else
      # 新サービスがまだなければモノリス内で処理
      legacy_notifications_action
    end
  end
 
  private
 
  def forward_to_new_service
    response = Faraday.new(NOTIFICATION_SERVICE_URL).send(
      request.method.downcase.to_sym,
      request.path,
      request.body.read,
      request.headers.select { |k, _| k.start_with?('HTTP_') }
    )
    render plain: response.body, status: response.status
  end
end

# notification-service/app/controllers/api/v1/notifications_controller.rb
class Api::V1::NotificationsController < ApplicationController
  before_action :authenticate_service!
 
  def create
    NotificationJob.perform_later(
      user_id: params[:user_id],
      type: params[:type],
      payload: params[:payload]
    )
    render json: { status: 'queued' }, status: :accepted
  end
 
  private
 
  def authenticate_service!
    token = request.headers['Authorization']&.split(' ')&.last
    head :unauthorized unless valid_service_token?(token)
  end
 
  def valid_service_token?(token)
    ActiveSupport::SecurityUtils.secure_compare(
      token.to_s,
      ENV.fetch('SERVICE_TOKEN')
    )
  end
end

# notification-service/app/jobs/notification_job.rb
class NotificationJob < ApplicationJob
  queue_as :notifications
  retry_on Net::SMTPServerBusy, wait: :exponentially_longer, attempts: 5
 
  def perform(user_id:, type:, payload:)
    user = UserFetcher.fetch(user_id) # Auth Serviceに問い合わせ
 
    case type
    when 'task_completed'
      NotificationMailer.task_completed(user, payload).deliver_now
    when 'mention'
      PushNotifier.send(user, payload)
    when 'deadline_reminder'
      NotificationMailer.deadline_reminder(user, payload).deliver_now
    end
  end
end

# app/services/notification_service.rb
class NotificationService
  BASE_URL = ENV['NOTIFICATION_SERVICE_URL']
 
  def self.send(user_id:, type:, payload: {})
    HTTParty.post(
      "#{BASE_URL}/api/v1/notifications",
      body: { user_id: user_id, type: type, payload: payload }.to_json,
      headers: {
        'Content-Type'  => 'application/json',
        'Authorization' => "Bearer #{Rails.application.credentials.notification_service_token}"
      },
      timeout: 5
    )
  rescue Net::ReadTimeout, Errno::ECONNREFUSED => e
    Rails.logger.error("Notification service error: #{e.message}")
    nil  # 通知サービスが落ちても、メインアプリは止めない
  end
end

# モノリスから通知関連を削除
$ git rm app/mailers/notification_mailer.rb
$ git rm app/workers/notification_worker.rb
$ git rm app/models/notification.rb
$ rails db:migrate  # notifications テーブルを削除
 
# 削除前後のコード量比較
Before: 847 Ruby files, 42891 lines
After:  791 Ruby files, 38642 lines  ← 通知関連4249行を削除

# Consumer側（task-service）: 期待するAPIの形を定義
# spec/pacts/task_notification_pact_spec.rb
describe "Task → Notification API Contract" do
  include Pact::Consumer::Minitest
 
  mock_service :notification_service do
    port 1234
  end
 
  it "sends task completed notification" do
    notification_service
      .upon_receiving("a task completion notification request")
      .with(
        method: :post,
        path: '/api/v1/notifications',
        body: {
          user_id: Pact.like(42),
          type: 'task_completed',
          payload: { task_id: Pact.like(100) }
        }
      )
      .will_respond_with(status: 202, body: { status: 'queued' })
 
    result = NotificationService.send(
      user_id: 42, type: 'task_completed',
      payload: { task_id: 100 }
    )
    assert_equal 202, result.code.to_i
  end
end

# Provider側（notification-service）: コントラクトを満たすか検証
Pact.service_provider "Notification Service" do
  honours_pact_with "Task Service" do
    pact_uri './pacts/task_service-notification_service.json'
  end
end

// proto/task/v1/task.proto
syntax = "proto3";
package task.v1;
option go_package = "github.com/echOtask/task-service/gen/task/v1";
 
service TaskService {
  rpc GetTask(GetTaskRequest) returns (GetTaskResponse);
  rpc CompleteTask(CompleteTaskRequest) returns (CompleteTaskResponse);
}
 
message Task {
  int64  id          = 1;
  string title       = 2;
  string status      = 3;
  int64  project_id  = 4;
  int64  assignee_id = 5;
}
 
message GetTaskRequest  { int64 task_id = 1; }
message GetTaskResponse { Task task = 1; }
message CompleteTaskRequest  { int64 task_id = 1; int64 user_id = 2; }
message CompleteTaskResponse { Task task = 1; }

// task-service/internal/server/task_server.go
package server
 
import (
	"context"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	taskv1 "github.com/echOtask/task-service/gen/task/v1"
	"github.com/echOtask/task-service/internal/repository"
)
 
type TaskServer struct {
	taskv1.UnimplementedTaskServiceServer
	repo repository.TaskRepository
}
 
func (s *TaskServer) GetTask(ctx context.Context, req *taskv1.GetTaskRequest) (*taskv1.GetTaskResponse, error) {
	task, err := s.repo.FindByID(ctx, req.TaskId)
	if err == repository.ErrNotFound {
		return nil, status.Errorf(codes.NotFound, "task %d not found", req.TaskId)
	}
	if err != nil {
		return nil, status.Errorf(codes.Internal, "internal error: %v", err)
	}
	return &taskv1.GetTaskResponse{Task: toProtoTask(task)}, nil
}
 
func (s *TaskServer) CompleteTask(ctx context.Context, req *taskv1.CompleteTaskRequest) (*taskv1.CompleteTaskResponse, error) {
	task, err := s.repo.Complete(ctx, req.TaskId, req.UserId)
	if err != nil {
		return nil, status.Errorf(codes.Internal, "failed: %v", err)
	}
	// 非同期でSNSにイベント発行（通知サービスが購読する）
	go s.publishTaskCompletedEvent(task)
	return &taskv1.CompleteTaskResponse{Task: toProtoTask(task)}, nil
}
 
// cmd/server/main.go: 認証・ロギング・リカバリのインターセプターチェーンでサーバー起動
func main() {
	lis, _ := net.Listen("tcp", ":50051")
	grpcServer := grpc.NewServer(
		grpc.ChainUnaryInterceptor(loggingInterceptor, recoveryInterceptor, authInterceptor),
	)
	taskv1.RegisterTaskServiceServer(grpcServer, &TaskServer{repo: repository.NewPostgres(initDB())})
	reflection.Register(grpcServer)
	log.Fatal(grpcServer.Serve(lis))
}

# タスクが完了したらSNSにイベントを発行
# task-service/app/models/task.rb
class Task < ApplicationRecord
  after_update :publish_event, if: :saved_change_to_status?
 
  private
 
  def publish_event
    EventPublisher.publish('task.updated', {
      task_id:         id,
      project_id:      project_id,
      status:          status,
      previous_status: status_before_last_save,
      occurred_at:     Time.current.iso8601
    })
  end
end
 
# task-service/app/services/event_publisher.rb
class EventPublisher
  def self.publish(event_type, payload)
    sns_client.publish(
      topic_arn: ENV.fetch('SNS_TOPIC_ARN'),
      message: payload.merge(event_type: event_type).to_json,
      message_attributes: {
        'event_type' => {
          data_type: 'String',
          string_value: event_type
        }
      }
    )
  end
 
  def self.sns_client
    @sns_client ||= Aws::SNS::Client.new(region: 'ap-northeast-1')
  end
  private_class_method :sns_client
end
 
# 通知サービスがSNS/SQSを通じてイベントを受け取る
# notification-service/app/workers/task_event_worker.rb
class TaskEventWorker
  include Shoryuken::Worker
 
  shoryuken_options queue: 'notification-service-tasks',
                    auto_delete: true,
                    body_parser: :json
 
  def perform(_sqs_msg, body)
    case body['event_type']
    when 'task.updated'
      handle_task_update(body)
    when 'task.completed'
      handle_task_completion(body)
    end
  end
 
  private
 
  def handle_task_completion(body)
    NotificationJob.perform_later(
      user_id: body['assignee_id'],
      type: 'task_completed',
      payload: body.slice('task_id', 'task_title')
    )
  end
end

# ECSサービスのDeploymentController設定
TaskECSService:
  Type: AWS::ECS::Service
  Properties:
    Cluster: !Ref ECSCluster
    DeploymentController:
      Type: CODE_DEPLOY          # CodeDeployによるBlue-Green制御
    LoadBalancers:
      - ContainerName: task-service
        ContainerPort: 3000
        TargetGroupArn: !Ref BlueTargetGroup  # 初期はBlueへ
 
# CodeDeployのデプロイグループ
TaskServiceDeployGroup:
  Type: AWS::CodeDeploy::DeploymentGroup
  Properties:
    DeploymentStyle:
      DeploymentType: BLUE_GREEN
      DeploymentOption: WITH_TRAFFIC_CONTROL
    BlueGreenDeploymentConfiguration:
      TerminateBlueInstancesOnDeploymentSuccess:
        Action: TERMINATE
        TerminationWaitTimeInMinutes: 5  # 旧バージョンを5分後に終了
    LoadBalancerInfo:
      TargetGroupPairInfoList:
        - ProdTrafficRoute:
            ListenerArns: [!Ref ALBProdListener]
          TargetGroups:
            - Name: !GetAtt BlueTargetGroup.TargetGroupName
            - Name: !GetAtt GreenTargetGroup.TargetGroupName

# まず10%のトラフィックを新バージョンへ、5分後に残り90%を切り替え
TaskServiceCanaryConfig:
  Type: AWS::CodeDeploy::DeploymentConfig
  Properties:
    DeploymentConfigName: task-service-canary-10
    ComputePlatform: ECS
    TrafficRoutingConfig:
      Type: TimeBasedCanary
      TimeBasedCanary:
        CanaryPercentage: 10
        CanaryInterval: 5

# ECSクラスターにService Connect名前空間を設定
EchoTaskCluster:
  Type: AWS::ECS::Cluster
  Properties:
    ClusterName: echOtask-production
    ServiceConnectDefaults:
      Namespace: echOtask.local  # 内部DNS名前空間
 
# Task ServiceにService Connect設定を追加
TaskECSService:
  Type: AWS::ECS::Service
  Properties:
    ServiceConnectConfiguration:
      Enabled: true
      Namespace: echOtask.local
      Services:
        - PortName: http
          DiscoveryName: task-service
          ClientAliases:
            - Port: 3000
              DnsName: task-service  # http://task-service:3000 でアクセス可能