oci-portal/internal/oci/logrelay.go

package oci

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"strings"
	"time"

	"github.com/oracle/oci-go-sdk/v65/common"
	"github.com/oracle/oci-go-sdk/v65/identity"
	"github.com/oracle/oci-go-sdk/v65/ons"
	"github.com/oracle/oci-go-sdk/v65/sch"
)

// 日志回传链路(方案A)在租户侧的固定资源命名,按名幂等查找与创建;
// 资源建在凭据默认区域,Audit 日志组含子 compartment。
// Topic 删除后名称有保留期(同名重建长时间 409 Conflict),故 Topic 用
// 前缀+随机后缀命名、按前缀幂等查找,销毁重建不受保留期阻塞。
const (
	relayTopicPrefix   = "ociportal-logs"
	relayPolicyName    = "ociportal-logs-sch"
	relayConnectorName = "ociportal-logs"
	relayAuditLogGroup = "_Audit_Include_Subcompartment"
	relayTopicPages    = 5 // 按前缀查找 Topic 的翻页上限
)

// Connector 创建为异步 work request,轮询直至 ACTIVE。
var (
	relayConnectorPollTick  = 5 * time.Second
	relayConnectorPollLimit = 18
)

// RelayResource 是链路单个云资源的状态;Created 标记本次调用新建,供失败回滚判据。
type RelayResource struct {
	ID      string `json:"id"`
	State   string `json:"state"`
	Created bool   `json:"-"`
}

// RelayState 是回传链路四资源现状;已删除的资源不出现(ID 为空)。
type RelayState struct {
	Topic        RelayResource `json:"topic"`
	Subscription RelayResource `json:"subscription"`
	Connector    RelayResource `json:"connector"`
	Policy       RelayResource `json:"policy"`
}

// RelayEventCondition 生成 Connector Log Filter 条件:按 eventName 白名单收窄回传。
func RelayEventCondition(events []string) string {
	terms := make([]string, 0, len(events))
	for _, e := range events {
		terms = append(terms, fmt.Sprintf("data.eventName='%s'", e))
	}
	return strings.Join(terms, " or ")
}

func (c *RealClient) onsControlClient(cred Credentials) (ons.NotificationControlPlaneClient, error) {
	cp, err := ons.NewNotificationControlPlaneClientWithConfigurationProvider(provider(cred))
	if err != nil {
		return cp, fmt.Errorf("new ons control client: %w", err)
	}
	applyProxy(&cp.BaseClient, cred)
	return cp, nil
}

func (c *RealClient) onsDataClient(cred Credentials) (ons.NotificationDataPlaneClient, error) {
	dp, err := ons.NewNotificationDataPlaneClientWithConfigurationProvider(provider(cred))
	if err != nil {
		return dp, fmt.Errorf("new ons data client: %w", err)
	}
	applyProxy(&dp.BaseClient, cred)
	return dp, nil
}

func (c *RealClient) schClient(cred Credentials) (sch.ServiceConnectorClient, error) {
	sc, err := sch.NewServiceConnectorClientWithConfigurationProvider(provider(cred))
	if err != nil {
		return sc, fmt.Errorf("new sch client: %w", err)
	}
	applyProxy(&sc.BaseClient, cred)
	return sc, nil
}

// EnsureRelayTopic 实现 Client:按前缀返回既有 Topic 或以随机后缀新建。
func (c *RealClient) EnsureRelayTopic(ctx context.Context, cred Credentials) (RelayResource, error) {
	cp, err := c.onsControlClient(cred)
	if err != nil {
		return RelayResource{}, err
	}
	if res, ok, err := findRelayTopic(ctx, cp, cred.TenancyOCID); err != nil || ok {
		return res, err
	}
	name, err := relayTopicNewName()
	if err != nil {
		return RelayResource{}, err
	}
	created, err := cp.CreateTopic(ctx, ons.CreateTopicRequest{
		CreateTopicDetails: ons.CreateTopicDetails{
			Name:          &name,
			CompartmentId: &cred.TenancyOCID,
			Description:   common.String("oci-portal 日志回传:关键审计事件经 Connector 投递到面板"),
		},
	})
	if err != nil {
		return RelayResource{}, fmt.Errorf("create ons topic: %w", err)
	}
	return RelayResource{ID: deref(created.TopicId), State: string(created.LifecycleState), Created: true}, nil
}

// relayTopicNewName 生成带随机后缀的 Topic 名,规避删除名称保留期。
func relayTopicNewName() (string, error) {
	buf := make([]byte, 4)
	if _, err := rand.Read(buf); err != nil {
		return "", fmt.Errorf("topic name suffix: %w", err)
	}
	return relayTopicPrefix + "-" + hex.EncodeToString(buf), nil
}

// findRelayTopic 按名称前缀查找存活 Topic;未找到时 ok 为 false。
func findRelayTopic(ctx context.Context, cp ons.NotificationControlPlaneClient, tenancy string) (RelayResource, bool, error) {
	req := ons.ListTopicsRequest{CompartmentId: &tenancy}
	for page := 0; page < relayTopicPages; page++ {
		list, err := cp.ListTopics(ctx, req)
		if err != nil {
			return RelayResource{}, false, fmt.Errorf("list ons topics: %w", err)
		}
		for _, t := range list.Items {
			if strings.HasPrefix(deref(t.Name), relayTopicPrefix) &&
				t.LifecycleState == ons.NotificationTopicSummaryLifecycleStateActive {
				return RelayResource{ID: deref(t.TopicId), State: string(t.LifecycleState)}, true, nil
			}
		}
		if list.OpcNextPage == nil {
			break
		}
		req.Page = list.OpcNextPage
	}
	return RelayResource{}, false, nil
}

// EnsureRelaySubscription 实现 Client:按 endpoint 返回既有 CUSTOM_HTTPS 订阅或新建;
// 新建订阅处于 PENDING,待 ONS 向 endpoint 投递确认消息、面板回访后转 ACTIVE。
func (c *RealClient) EnsureRelaySubscription(ctx context.Context, cred Credentials, topicID, endpoint string) (RelayResource, error) {
	dp, err := c.onsDataClient(cred)
	if err != nil {
		return RelayResource{}, err
	}
	if res, ok, err := findRelaySubscription(ctx, dp, cred.TenancyOCID, topicID, endpoint); err != nil || ok {
		return res, err
	}
	created, err := dp.CreateSubscription(ctx, ons.CreateSubscriptionRequest{
		CreateSubscriptionDetails: ons.CreateSubscriptionDetails{
			TopicId:       &topicID,
			CompartmentId: &cred.TenancyOCID,
			Protocol:      common.String("CUSTOM_HTTPS"),
			Endpoint:      &endpoint,
		},
	})
	if err != nil {
		return RelayResource{}, fmt.Errorf("create ons subscription: %w", err)
	}
	return RelayResource{ID: deref(created.Id), State: string(created.LifecycleState), Created: true}, nil
}

// findRelaySubscription 在 Topic 下按 endpoint 查找存活订阅。
func findRelaySubscription(ctx context.Context, dp ons.NotificationDataPlaneClient, tenancy, topicID, endpoint string) (RelayResource, bool, error) {
	if endpoint == "" {
		return RelayResource{}, false, nil
	}
	list, err := dp.ListSubscriptions(ctx, ons.ListSubscriptionsRequest{
		CompartmentId: &tenancy, TopicId: &topicID,
	})
	if err != nil {
		return RelayResource{}, false, fmt.Errorf("list ons subscriptions: %w", err)
	}
	for _, s := range list.Items {
		if deref(s.Endpoint) == endpoint && s.LifecycleState != ons.SubscriptionSummaryLifecycleStateDeleted {
			return RelayResource{ID: deref(s.Id), State: string(s.LifecycleState)}, true, nil
		}
	}
	return RelayResource{}, false, nil
}

// GetRelaySubscription 实现 Client:查询订阅当前状态,供确认期轮询。
func (c *RealClient) GetRelaySubscription(ctx context.Context, cred Credentials, subscriptionID string) (RelayResource, error) {
	dp, err := c.onsDataClient(cred)
	if err != nil {
		return RelayResource{}, err
	}
	resp, err := dp.GetSubscription(ctx, ons.GetSubscriptionRequest{SubscriptionId: &subscriptionID})
	if err != nil {
		return RelayResource{}, fmt.Errorf("get ons subscription: %w", err)
	}
	return RelayResource{ID: deref(resp.Id), State: string(resp.LifecycleState)}, nil
}

// EnsureRelayPolicy 实现 Client:按名返回既有 IAM policy 或新建;
// 写操作必须发往 home region,授权 Service Connector 发布消息到 Topic。
func (c *RealClient) EnsureRelayPolicy(ctx context.Context, cred Credentials, homeRegion string) (RelayResource, error) {
	ic, err := c.identityClientAt(cred, homeRegion)
	if err != nil {
		return RelayResource{}, err
	}
	list, err := ic.ListPolicies(ctx, identity.ListPoliciesRequest{
		CompartmentId: &cred.TenancyOCID, Name: common.String(relayPolicyName),
	})
	if err != nil {
		return RelayResource{}, fmt.Errorf("list policies: %w", err)
	}
	if len(list.Items) > 0 {
		return RelayResource{ID: deref(list.Items[0].Id), State: string(list.Items[0].LifecycleState)}, nil
	}
	stmt := fmt.Sprintf("Allow any-user to use ons-topics in tenancy where all {request.principal.type='serviceconnector', request.principal.compartment.id='%s'}", cred.TenancyOCID)
	created, err := ic.CreatePolicy(ctx, identity.CreatePolicyRequest{
		CreatePolicyDetails: identity.CreatePolicyDetails{
			CompartmentId: &cred.TenancyOCID,
			Name:          common.String(relayPolicyName),
			Description:   common.String("oci-portal 日志回传:允许 Service Connector 发布到 ONS Topic"),
			Statements:    []string{stmt},
		},
	})
	if err != nil {
		return RelayResource{}, fmt.Errorf("create policy: %w", err)
	}
	return RelayResource{ID: deref(created.Id), State: string(created.LifecycleState), Created: true}, nil
}

// EnsureRelayConnector 实现 Client:按名返回既有 Connector 或新建(_Audit 含子区间 → Topic,
// 按 condition 过滤);新建后轮询至 ACTIVE,超时返回错误但保留 Created 供上层回滚。
func (c *RealClient) EnsureRelayConnector(ctx context.Context, cred Credentials, topicID, condition string) (RelayResource, error) {
	sc, err := c.schClient(cred)
	if err != nil {
		return RelayResource{}, err
	}
	if res, ok, err := findRelayConnector(ctx, sc, cred.TenancyOCID); err != nil || ok {
		return res, err
	}
	details := sch.CreateServiceConnectorDetails{
		DisplayName:   common.String(relayConnectorName),
		CompartmentId: &cred.TenancyOCID,
		Source: sch.LoggingSourceDetails{LogSources: []sch.LogSource{{
			CompartmentId: &cred.TenancyOCID,
			LogGroupId:    common.String(relayAuditLogGroup),
		}}},
		Target: sch.NotificationsTargetDetails{TopicId: &topicID},
	}
	if condition != "" {
		details.Tasks = []sch.TaskDetails{sch.LogRuleTaskDetails{Condition: &condition}}
	}
	if _, err := sc.CreateServiceConnector(ctx, sch.CreateServiceConnectorRequest{
		CreateServiceConnectorDetails: details,
	}); err != nil {
		return RelayResource{}, fmt.Errorf("create service connector: %w", err)
	}
	return waitRelayConnector(ctx, sc, cred.TenancyOCID)
}

// findRelayConnector 按名查找存活 Connector。
func findRelayConnector(ctx context.Context, sc sch.ServiceConnectorClient, tenancy string) (RelayResource, bool, error) {
	list, err := sc.ListServiceConnectors(ctx, sch.ListServiceConnectorsRequest{
		CompartmentId: &tenancy, DisplayName: common.String(relayConnectorName),
	})
	if err != nil {
		return RelayResource{}, false, fmt.Errorf("list service connectors: %w", err)
	}
	for _, item := range list.Items {
		if item.LifecycleState != sch.LifecycleStateDeleted && item.LifecycleState != sch.LifecycleStateDeleting {
			return RelayResource{ID: deref(item.Id), State: string(item.LifecycleState)}, true, nil
		}
	}
	return RelayResource{}, false, nil
}

// waitRelayConnector 轮询新建 Connector 直至 ACTIVE;超时带回已建资源信息。
func waitRelayConnector(ctx context.Context, sc sch.ServiceConnectorClient, tenancy string) (RelayResource, error) {
	last := RelayResource{Created: true}
	for i := 0; i < relayConnectorPollLimit; i++ {
		select {
		case <-ctx.Done():
			return last, ctx.Err()
		case <-time.After(relayConnectorPollTick):
		}
		res, ok, err := findRelayConnector(ctx, sc, tenancy)
		if err != nil {
			return last, err
		}
		if ok {
			last, last.Created = res, true
			if res.State == string(sch.LifecycleStateActive) {
				return last, nil
			}
		}
	}
	return last, fmt.Errorf("service connector not active after %s", time.Duration(relayConnectorPollLimit)*relayConnectorPollTick)
}

// RelayState 实现 Client:聚合链路四资源现状;endpoint 为空时不匹配订阅。
func (c *RealClient) RelayState(ctx context.Context, cred Credentials, endpoint string) (RelayState, error) {
	var st RelayState
	cp, err := c.onsControlClient(cred)
	if err != nil {
		return st, err
	}
	if st.Topic, _, err = findRelayTopic(ctx, cp, cred.TenancyOCID); err != nil {
		return st, err
	}
	if st.Topic.ID != "" {
		dp, err := c.onsDataClient(cred)
		if err != nil {
			return st, err
		}
		if st.Subscription, _, err = findRelaySubscription(ctx, dp, cred.TenancyOCID, st.Topic.ID, endpoint); err != nil {
			return st, err
		}
	}
	return c.relayControlState(ctx, cred, st)
}

// relayControlState 补齐 Connector 与 Policy 两项状态。
func (c *RealClient) relayControlState(ctx context.Context, cred Credentials, st RelayState) (RelayState, error) {
	sc, err := c.schClient(cred)
	if err != nil {
		return st, err
	}
	if st.Connector, _, err = findRelayConnector(ctx, sc, cred.TenancyOCID); err != nil {
		return st, err
	}
	ic, err := c.identityClientAt(cred, "")
	if err != nil {
		return st, err
	}
	list, err := ic.ListPolicies(ctx, identity.ListPoliciesRequest{
		CompartmentId: &cred.TenancyOCID, Name: common.String(relayPolicyName),
	})
	if err != nil {
		return st, fmt.Errorf("list policies: %w", err)
	}
	if len(list.Items) > 0 {
		st.Policy = RelayResource{ID: deref(list.Items[0].Id), State: string(list.Items[0].LifecycleState)}
	}
	return st, nil
}

// DeleteRelayConnector 实现 Client。
func (c *RealClient) DeleteRelayConnector(ctx context.Context, cred Credentials, connectorID string) error {
	sc, err := c.schClient(cred)
	if err != nil {
		return err
	}
	if _, err := sc.DeleteServiceConnector(ctx, sch.DeleteServiceConnectorRequest{ServiceConnectorId: &connectorID}); err != nil {
		return fmt.Errorf("delete service connector: %w", err)
	}
	return nil
}

// DeleteRelayPolicy 实现 Client;IAM 写操作发往 home region。
func (c *RealClient) DeleteRelayPolicy(ctx context.Context, cred Credentials, homeRegion, policyID string) error {
	ic, err := c.identityClientAt(cred, homeRegion)
	if err != nil {
		return err
	}
	if _, err := ic.DeletePolicy(ctx, identity.DeletePolicyRequest{PolicyId: &policyID}); err != nil {
		return fmt.Errorf("delete policy: %w", err)
	}
	return nil
}

// DeleteRelaySubscription 实现 Client。
func (c *RealClient) DeleteRelaySubscription(ctx context.Context, cred Credentials, subscriptionID string) error {
	dp, err := c.onsDataClient(cred)
	if err != nil {
		return err
	}
	if _, err := dp.DeleteSubscription(ctx, ons.DeleteSubscriptionRequest{SubscriptionId: &subscriptionID}); err != nil {
		return fmt.Errorf("delete ons subscription: %w", err)
	}
	return nil
}

// DeleteRelayTopic 实现 Client。
func (c *RealClient) DeleteRelayTopic(ctx context.Context, cred Credentials, topicID string) error {
	cp, err := c.onsControlClient(cred)
	if err != nil {
		return err
	}
	if _, err := cp.DeleteTopic(ctx, ons.DeleteTopicRequest{TopicId: &topicID}); err != nil {
		return fmt.Errorf("delete ons topic: %w", err)
	}
	return nil
}