弁護士ドットコム株式会社 Creators’ blog

弁護士ドットコムがエンジニア・デザイナーのサービス開発事例やデザイン活動を発信する公式ブログです。

OpenFeature で実現するベンダーフリーな機能フラグ管理

バックエンド リーガルブレイン開発 アドベントカレンダー

この記事は弁護士ドットコム Advent Calendar 2024 の 24 日目の記事です。

弁護士ドットコム株式会社リーガルブレイン開発室の伊藤です。

リーガルブレイン開発室では、あらゆる法律関連データと生成 AI を組み合わせたプロダクトを開発しています。そのうちのひとつである「チャット法律相談」は、先日開催された生成AI大賞2024 1 にて特別賞を受賞しました。

www.bengo4.com

直近では、これまでの開発で得られた知見を活かして統合型 AI リーガルリサーチツールの開発を進めています。

www.bengo4.com

このような新サービスの開発または新機能の開発においては、機能を段階的にリリースしたり複数のパターンの実装に対するユーザーの反応を比較したくなる場面が多々あります。そのような場面で有用なのが機能フラグ 2 です。

機能フラグについては、昨年のアドベントカレンダーで Azure App Configuration を使った機能の切り替えについて紹介しましたので、そちらもご一読ください。

creators.bengo4.com

今回は、機能フラグの標準仕様である OpenFeature の概念を取り入れることで、特定のベンダーに依存しない機能フラグ管理を実現する方法について紹介します。

OpenFeature について

OpenFeature3 は、特定のベンダーに依存しない機能フラグ管理を実現するための API を提供する、コミュニティ主導のオープン仕様です。機能フラグ管理が標準化されることで、機能フラグを利用するツール・アプリケーションと、機能フラグを提供するベンダーとが共通のインタフェースを通じて結合されます。その結果、コードレベルでのベンダーロックインを回避できます。

例えば、App Configuration を使って機能フラグ管理をある日突然 LaunchDarkly4 に変更したくなった場合でも、アプリケーションのコード変更を最小限に抑えることができます。

この記事の中でやりたいこと

まず前半では、App Configuration で管理する機能フラグを OpenFeature のインタフェースを通じて利用できるようにするためのプロバイダーを Python で実装します。実装したプロバイダーを使用すると、機能フラグは下記のようなイメージで評価できます。

OpenFeature のインタフェースを利用した機能フラグ評価のイメージをシーケンス図で表したもの

そして後半では、OpenFeature の特徴であるベンダーフリーであることを感じていただきます。上のシーケンスを見ていただくと、アプリケーションでは下記の事柄だけ考えればよいことがわかります。

  • どのプロバイダーを使うか
  • OpenFeature の EvaluationAPI を呼ぶ

この記事では、実際に最小限のコード変更により機能フラグの管理を App Configuration から LaunchDarkly に切り替える方法を紹介します。

App Configuration を OpenFeature のプロバイダーとして実装する

まずは App Configuration を OpenFeature のインタフェースを通じて利用するためのプロバイダーを実装します。ただし、今回は真偽値をとる機能フラグを評価するためだけの簡易的な実装とします。

OpenFeature のプロバイダーとして必要な実装

Python で OpenFeature のプロバイダーを実装する場合、OpenFeature の python-sdk にある AbstractFeatureProvider を継承して class を実装します。

python-sdk/openfeature/provider/init.py at main · open-feature/python-sdk

AbstractFeatureProvider の実装は上記リンク先を参照していただきたいのですが、この class を継承する際、最低でも下記のメソッドを実装する必要があります。

def get_metadata(self) -> Metadata:
    pass

def get_provider_hooks(self) -> typing.List[Hook]:
    return []

def resolve_boolean_details(
    self,
    flag_key: str,
    default_value: bool,
    evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[bool]:
    pass

def resolve_string_details(
    self,
    flag_key: str,
    default_value: str,
    evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[str]:
    pass

def resolve_integer_details(
    self,
    flag_key: str,
    default_value: int,
    evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[int]:
    pass

def resolve_float_details(
    self,
    flag_key: str,
    default_value: float,
    evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[float]:
    pass

def resolve_object_details(
    self,
    flag_key: str,
    default_value: typing.Union[dict, list],
    evaluation_context: typing.Optional[EvaluationContext] = None,
) -> FlagResolutionDetails[typing.Union[dict, list]]:
    pass

次のステップでは、AbstractFeatureProvider を継承して App Configuration を利用するプロバイダーを実装します。

App Configuration のプロバイダーの実装

class Config:
    def __init__(self, endpoint: str, credential: TokenCredential):
        self.endpoint = endpoint
        self.credential = credential

class AppConfigurationProvider(AbstractProvider):
    def __init__(self, config: Config):
        self.config = config

    def get_metadata(self) -> Metadata:
        return Metadata("azure-app-configuration-server")

    def get_provider_hooks(self) -> List[Hook]:
        return []

    def resolve_boolean_details(
        self,
        flag_key: str,
        default_value: bool,
        evaluation_context: EvaluationContext | None,
    ) -> FlagResolutionDetails[bool]:
        return self.__resolve_value(FlagType.BOOLEAN, flag_key, default_value, evaluation_context)

    def resolve_string_details(
        self,
        flag_key: str,
        default_value: str,
        evaluation_context: EvaluationContext | None,
    ) -> FlagResolutionDetails[str]:
        return FlagResolutionDetails(value=default_value, reason=Reason(Reason.ERROR), error_code=ErrorCode.TYPE_MISMATCH)

    def resolve_integer_details(
        self,
        flag_key: str,
        default_value: int,
        evaluation_context: EvaluationContext | None,
    ) -> FlagResolutionDetails[int]:
        return FlagResolutionDetails(value=default_value, reason=Reason(Reason.ERROR), error_code=ErrorCode.TYPE_MISMATCH)

    def resolve_float_details(
        self,
        flag_key: str,
        default_value: float,
        evaluation_context: EvaluationContext | None,
    ) -> FlagResolutionDetails[float]:
        return FlagResolutionDetails(value=default_value, reason=Reason(Reason.ERROR), error_code=ErrorCode.TYPE_MISMATCH)

    def resolve_object_details(
        self,
        flag_key: str,
        default_value: Union[dict, list],
        evaluation_context: EvaluationContext | None,
    ) -> FlagResolutionDetails[Union[dict, list]]:
        return FlagResolutionDetails(value=default_value, reason=Reason(Reason.ERROR), error_code=ErrorCode.TYPE_MISMATCH)

    def __resolve_value(
        self,
        flag_type: FlagType,
        flag_key: str,
        default_value: Any,
        evaluation_context: EvaluationContext | None = None,
    ) -> FlagResolutionDetails:
        if evaluation_context is None:
            return FlagResolutionDetails(
                value=default_value,
                reason=Reason(Reason.ERROR),
                error_code=ErrorCode.TARGETING_KEY_MISSING,
            )

        try:
            config_setting = load(
                endpoint=self.config.endpoint,
                credential=self.config.credential,
                feature_flag_enabled=True,
                feature_flag_refresh_enabled=True,
            )

            if config_setting is None:
                return FlagResolutionDetails(
                    value=default_value,
                    reason=Reason(Reason.ERROR),
                    error_code=ErrorCode.FLAG_NOT_FOUND,
                )

            feature_manager = FeatureManager(config_setting)
            enabled = feature_manager.is_enabled(
                flag_key, TargetingContext(user_id=evaluation_context.targeting_key)
            )

            return FlagResolutionDetails(
                value=enabled,
                reason=(Reason.TARGETING_MATCH if enabled else Reason.DISABLED),
            )

        except Exception:
            return FlagResolutionDetails(
                value=default_value,
                reason=Reason(Reason.ERROR),
                error_code=ErrorCode.PROVIDER_NOT_READY,
                error_message=f"Error occurred while resolving flag value. {traceback.format_exc()}",
            )

def init_provider() -> AppConfigurationProvider:
    return AppConfigurationProvider(
        Config(
            endpoint=environ.get("APP_CONFIG_ENDPOINT"),
            credential=DefaultAzureCredential(),
        )
    )

肝になる部分は、実際に App Configuration から機能フラグを取得して評価する __resolve_value メソッドです。このメソッドでは、App Configuration から取得した機能フラグ設定に対して、OpenFeature の概念である EvaluationContext として渡されたユーザー ID を App Configuration の概念である TargetingContext として利用して機能フラグを評価しています。また機能フラグの評価をするために microsoft/FeatureManagement-Python で提供されている FeatureManager を利用しています。

機能フラグの評価に関する詳細な実装については、Microsoft の公式ドキュメントを参照してください。

Quickstart for adding feature flags to Python with App Configuration | Microsoft Learn

プロバイダーを使用して機能フラグを評価する

では、実装したプロバイダーを使用して機能フラグを評価してみます。

評価用の機能フラグおよび App Configuration のリソースは、あらかじめ Terraform で作成しておきます。

terraform {
  required_version = ">= 1.10.1"

  required_providers {
    azurerm = {
      source  = "hashicorp/azurerm"
      version = "=4.13.0"
    }
  }
}

provider "azurerm" {
  subscription_id = var.azure_subscription_id

  features {}
}

resource "azurerm_resource_group" "main" {
  name     = "${local.project_name}-rg"
  location = var.azure_location
}

resource "azurerm_app_configuration" "main" {
  name                = "${local.project_name}-appconfig"
  resource_group_name = azurerm_resource_group.main.name
  location            = azurerm_resource_group.main.location
}

data "azurerm_client_config" "current" {}

resource "azurerm_role_assignment" "appconfig_dataowner" {
  scope                = azurerm_app_configuration.main.id
  role_definition_name = "App Configuration Data Owner"
  principal_id         = data.azurerm_client_config.current.object_id
}

resource "azurerm_app_configuration_feature" "feature" {
  configuration_store_id = azurerm_app_configuration.main.id
  name                   = local.feature_name
  description            = "Feature flag for ${local.feature_name}"
  enabled                = true
  targeting_filter {
    default_rollout_percentage = 0
    users                      = ["user-1"]
  }

  depends_on = [
    azurerm_role_assignment.appconfig_dataowner,
  ]
}

output "app_config_endpoint" {
  value = azurerm_app_configuration.main.endpoint
}

機能フラグを評価するためのスクリプトを以下に示します。

import sys
from pprint import pprint as pp

from openfeature import api
from openfeature.client import OpenFeatureClient
from openfeature.evaluation_context import EvaluationContext

import appconfig as ac

FEATURE_NAME: str = "is_sushi"


def main(user_id: str) -> int:
    provider = ac.init_provider()

    api.set_provider(provider)

    client: OpenFeatureClient = api.get_client()

    detail = client.get_boolean_details(
        flag_key=FEATURE_NAME,
        default_value=False,
        evaluation_context=EvaluationContext(
            targeting_key=user_id,
            attributes={},
        ),
    )

    pp(detail)

    return 0


if __name__ == "__main__":
    user_id: str = "user-0"

    args = sys.argv
    if len(args) > 1:
        user_id = args[1]

    sys.exit(main(user_id=user_id))

例えば、user-0 で評価する場合の実行結果は下記のようになります。

❯ python main.py 'user-0'

FlagEvaluationDetails(flag_key='is_sushi',
                      value=False,
                      variant=None,
                      flag_metadata={},
                      reason=<Reason.DISABLED: 'DISABLED'>,
                      error_code=None,
                      error_message=None)

そして、user-1 で評価する場合の実行結果は下記のようになります。

❯ python main.py 'user-1'

FlagEvaluationDetails(flag_key='is_sushi',
                      value=True,
                      variant=None,
                      flag_metadata={},
                      reason=<Reason.TARGETING_MATCH: 'TARGETING_MATCH'>,
                      error_code=None,
                      error_message=None)

機能フラグ管理を LaunchDarkly に切り替える

最後に、実装したプロバイダーを使用して機能フラグを評価するアプリケーションを LaunchDarkly に切り替えてみます。

まずは App Configuration に作成した機能フラグと同じ名前の機能フラグを LaunchDarkly にも作成します。 こちらも Terraform で作成します。

terraform {
  required_version = ">= 1.10.1"

  required_providers {
    launchdarkly = {
      source  = "launchdarkly/launchdarkly"
      version = "~> 2.0"
    }
  }
}

provider "launchdarkly" {
  access_token = var.launchdarkly_access_token
}

resource "launchdarkly_project" "main" {
  key  = local.project_name
  name = local.project_name

  environments {
    key   = "test"
    name  = "Test"
    color = "F5A623"
  }

  tags = [
    "terraform",
    "test",
  ]
}


resource "launchdarkly_feature_flag" "main" {
  project_key = launchdarkly_project.main.key
  key         = local.feature_name
  name        = local.feature_name
  description = "Feature flag for ${local.feature_name}"

  variation_type = "boolean"

  variations {
    value = false
    name  = "Disabled"
  }
  variations {
    value = true
    name  = "Enabled"
  }

  defaults {
    on_variation  = 0
    off_variation = 1
  }

  tags = ["terraform", "test"]
}

resource "launchdarkly_feature_flag_environment" "main" {
  flag_id = launchdarkly_feature_flag.main.id
  env_key = launchdarkly_project.main.environments[0].key

  on = true

  targets {
    values    = ["user-1"]
    variation = 1
  }

  fallthrough {
    variation = 0
  }

  off_variation = 0
}

LaunchDarkly 用の OpenFeature プロバイダーはすでに公開されているものがある5 ため、プロバイダーの初期化をする関数だけ実装しておきます。

from os import environ

from ld_openfeature import LaunchDarklyProvider
from ldclient import Config


def init_provider():
    sdk_key: str = environ.get("LD_SDK_KEY")
    return LaunchDarklyProvider(Config(sdk_key))

そして機能フラグ評価用のスクリプトを変更するのですが、なんと変更箇所はプロバイダーの初期化関数のみです。

@@ -5,13 +5,13 @@
 from openfeature.client import OpenFeatureClient
 from openfeature.evaluation_context import EvaluationContext
 
-import appconfig as ac
+import launchdarkly as ld
 
 FEATURE_NAME: str = "is_sushi"
 
 
 def main(user_id: str) -> int:
-    provider = ac.init_provider()
+    provider = ld.init_provider()
 
     api.set_provider(provider)

実際に user-0 と user-1 で実行してみた結果は下記のとおりです。

❯ python main.py 'user-0'

FlagEvaluationDetails(flag_key='is_sushi',
                      value=False,
                      variant='0',
                      flag_metadata={},
                      reason='FALLTHROUGH',
                      error_code=None,
                      error_message=None)


❯ python main.py 'user-1'

FlagEvaluationDetails(flag_key='is_sushi',
                      value=True,
                      variant='1',
                      flag_metadata={},
                      reason=<Reason.TARGETING_MATCH: 'TARGETING_MATCH'>,
                      error_code=None,
                      error_message=None)

プロバイダーの実装の関係で評価理由 (reason) の値は異なっていますが、評価結果は一致しています。

このように、最小限のコード変更により機能フラグ管理を App Configuration から LaunchDarkly に切り替えることができました。

まとめ

この記事では、OpenFeature の概念を取り入れることで、特定のベンダーに依存しない機能フラグ管理を実現する方法について紹介しました。

App Configuration と LaunchDarkly を例に挙げましたが、他の機能フラグ管理サービスでも同様の方法でベンダーフリーに機能フラグを管理できます。

今回紹介したプロバイダーの実装と Terraform のコードについては下記のリンク先で公開しています。

github.com

App Configuration、LaunchDarkly ともに無料の範囲内で利用できるので、ぜひ試してみてください。

  1. 生成AI大賞2024↩
  2. 同様の機能を指す用語として "フィーチャーフラグ" や "リリースフラグ" などがありますが、この記事内では "機能フラグ" と統一して表記します。↩
  3. OpenFeature↩
  4. LaunchDarkly: Feature Flags, Feature Management, and Experimentation↩
  5. launchdarkly/openfeature-python-server: An OpenFeature provider for the LaunchDarkly Python server SDK.↩

© Bengo4.com, Inc. 2005