feat: resource log

CHANGELOG.md

@@ -6,6 +6,7 @@
 1. [16547](https://github.com/influxdata/influxdb/pull/16547): Allow trailing newline in credentials file and CLI integration
 1. [16545](https://github.com/influxdata/influxdb/pull/16545): Add support for prefixed cursor search to ForwardCursor types
 1. [16504](https://github.com/influxdata/influxdb/pull/16504): Add backup and restore
+1. [16522](https://github.com/influxdata/influxdb/pull/16522): Introduce resource logger to tasks, buckets and organizations
 
 ### UI Improvements
 

bucket.go

@@ -2,6 +2,7 @@ package influxdb
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 )
@@ -68,6 +69,7 @@ var (
 	OpFindBucket     = "FindBucket"
 	OpFindBuckets    = "FindBuckets"
 	OpCreateBucket   = "CreateBucket"
+	OpPutBucket      = "PutBucket"
 	OpUpdateBucket   = "UpdateBucket"
 	OpDeleteBucket   = "DeleteBucket"
 )
@@ -153,3 +155,12 @@ func (f BucketFilter) String() string {
 	}
 	return "[" + strings.Join(parts, ", ") + "]"
 }
+
+func ErrInternalBucketServiceError(op string, err error) *Error {
+	return &Error{
+		Code: EInternal,
+		Msg:  fmt.Sprintf("unexpected error in buckets; Err: %v", err),
+		Op:   op,
+		Err:  err,
+	}
+}

context/token.go

@@ -57,3 +57,12 @@ func GetToken(ctx context.Context) (string, error) {
 
 	return auth.Token, nil
 }
+
+// GetUserID retrieves the user ID from the authorizer on the context.
+func GetUserID(ctx context.Context) (influxdb.ID, error) {
+	a, err := GetAuthorizer(ctx)
+	if err != nil {
+		return 0, err
+	}
+	return a.GetUserID(), nil
+}

context/token_test.go

@@ -8,6 +8,21 @@ import (
 	icontext "github.com/influxdata/influxdb/context"
 )
 
+func TestGetAuthorizer(t *testing.T) {
+	ctx := context.Background()
+	ctx = icontext.SetAuthorizer(ctx, &influxdb.Authorization{
+		ID: 1234,
+	})
+	got, err := icontext.GetAuthorizer(ctx)
+	if err != nil {
+		t.Errorf("unexpected error while retrieving token: %v", err)
+	}
+
+	if want := influxdb.ID(1234); got.Identifier() != want {
+		t.Errorf("GetToken() want %s, got %s", want, got)
+	}
+}
+
 func TestGetToken(t *testing.T) {
 	ctx := context.Background()
 	ctx = icontext.SetAuthorizer(ctx, &influxdb.Authorization{
@@ -22,3 +37,18 @@ func TestGetToken(t *testing.T) {
 		t.Errorf("GetToken() want %s, got %s", want, got)
 	}
 }
+
+func TestGetUserID(t *testing.T) {
+	ctx := context.Background()
+	ctx = icontext.SetAuthorizer(ctx, &influxdb.Authorization{
+		UserID: 5678,
+	})
+	got, err := icontext.GetUserID(ctx)
+	if err != nil {
+		t.Errorf("unexpected error while retrieving user ID: %v", err)
+	}
+
+	if want := influxdb.ID(5678); got != want {
+		t.Errorf("GetUserID() want %s, got %s", want, got)
+	}
+}

kv/bucket.go

@@ -9,6 +9,7 @@ import (
 	"github.com/influxdata/influxdb"
 	icontext "github.com/influxdata/influxdb/context"
 	"github.com/influxdata/influxdb/kit/tracing"
+	"github.com/influxdata/influxdb/resource"
 )
 
 var (
@@ -477,14 +478,28 @@ func (s *Service) createBucket(ctx context.Context, tx Tx, b *influxdb.Bucket) (
 		}
 	}
 
-	if err := s.putBucket(ctx, tx, b); err != nil {
+	v, err := json.Marshal(b)
+	if err != nil {
+		return influxdb.ErrInternalBucketServiceError(influxdb.OpCreateBucket, err)
+	}
+	if err := s.putBucket(ctx, tx, b, v); err != nil {
 		return err
 	}
 
 	if err := s.createUserResourceMappingForOrg(ctx, tx, b.OrgID, b.ID, influxdb.BucketsResourceType); err != nil {
 		return err
 	}
-	return nil
+
+	uid, _ := icontext.GetUserID(ctx)
+	return s.audit.Log(resource.Change{
+		Type:           resource.Create,
+		ResourceID:     b.ID,
+		ResourceType:   influxdb.BucketsResourceType,
+		OrganizationID: b.OrgID,
+		UserID:         uid,
+		ResourceBody:   v,
+		Time:           time.Now(),
+	})
 }
 
 func (s *Service) generateBucketID(ctx context.Context, tx Tx) (influxdb.ID, error) {
@@ -494,21 +509,32 @@ func (s *Service) generateBucketID(ctx context.Context, tx Tx) (influxdb.ID, err
 // PutBucket will put a bucket without setting an ID.
 func (s *Service) PutBucket(ctx context.Context, b *influxdb.Bucket) error {
 	return s.kv.Update(ctx, func(tx Tx) error {
-		return s.putBucket(ctx, tx, b)
+		v, err := json.Marshal(b)
+		if err != nil {
+			return influxdb.ErrInternalBucketServiceError(influxdb.OpPutBucket, err)
+		}
+
+		if err := s.putBucket(ctx, tx, b, v); err != nil {
+			return err
+		}
+
+		uid, _ := icontext.GetUserID(ctx)
+		return s.audit.Log(resource.Change{
+			Type:           resource.Put,
+			ResourceID:     b.ID,
+			ResourceType:   influxdb.BucketsResourceType,
+			OrganizationID: b.OrgID,
+			UserID:         uid,
+			ResourceBody:   v,
+			Time:           time.Now(),
+		})
 	})
 }
 
-func (s *Service) putBucket(ctx context.Context, tx Tx, b *influxdb.Bucket) error {
+func (s *Service) putBucket(ctx context.Context, tx Tx, b *influxdb.Bucket, v []byte) error {
 	span, _ := tracing.StartSpanFromContext(ctx)
 	defer span.Finish()
 
-	v, err := json.Marshal(b)
-	if err != nil {
-		return &influxdb.Error{
-			Err: err,
-		}
-	}
-
 	encodedID, err := b.ID.Encode()
 	if err != nil {
 		return &influxdb.Error{
@@ -682,18 +708,36 @@ func (s *Service) updateBucket(ctx context.Context, tx Tx, id influxdb.ID, upd i
 		return nil, err
 	}
 
-	if err := s.putBucket(ctx, tx, b); err != nil {
+	v, err := json.Marshal(b)
+	if err != nil {
+		return nil, influxdb.ErrInternalBucketServiceError(influxdb.OpUpdateBucket, err)
+	}
+
+	if err := s.putBucket(ctx, tx, b, v); err != nil {
 		return nil, err
 	}
 
+	uid, _ := icontext.GetUserID(ctx)
+	if err := s.audit.Log(resource.Change{
+		Type:           resource.Update,
+		ResourceID:     b.ID,
+		ResourceType:   influxdb.BucketsResourceType,
+		OrganizationID: b.OrgID,
+		UserID:         uid,
+		ResourceBody:   v,
+		Time:           time.Now(),
+	}); err != nil {
+		return nil, &influxdb.Error{
+			Err: err,
+		}
+	}
+
 	return b, nil
 }
 
 // DeleteBucket deletes a bucket and prunes it from the index.
 func (s *Service) DeleteBucket(ctx context.Context, id influxdb.ID) error {
 	return s.kv.Update(ctx, func(tx Tx) error {
-		var err error
-
 		bucket, err := s.findBucketByID(ctx, tx, id)
 		if err != nil && !IsNotFound(err) {
 			return err
@@ -706,10 +750,19 @@ func (s *Service) DeleteBucket(ctx context.Context, id influxdb.ID) error {
 			}
 		}
 
-		if pe := s.deleteBucket(ctx, tx, id); pe != nil {
-			err = pe
+		if err := s.deleteBucket(ctx, tx, id); err != nil {
+			return err
 		}
-		return err
+
+		uid, _ := icontext.GetUserID(ctx)
+		return s.audit.Log(resource.Change{
+			Type:           resource.Delete,
+			ResourceID:     id,
+			ResourceType:   influxdb.BucketsResourceType,
+			OrganizationID: bucket.OrgID,
+			UserID:         uid,
+			Time:           time.Now(),
+		})
 	})
 }
 

kv/org.go

@@ -7,6 +7,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/influxdata/influxdb/resource"
+
 	"go.uber.org/zap"
 
 	"github.com/influxdata/influxdb"
@@ -293,12 +295,26 @@ func (s *Service) createOrganization(ctx context.Context, tx Tx, o *influxdb.Org
 		}
 	}
 
-	if err := s.putOrganization(ctx, tx, o); err != nil {
+	v, err := json.Marshal(o)
+	if err != nil {
+		return influxdb.ErrInternalOrgServiceError(influxdb.OpCreateOrganization, err)
+	}
+	if err := s.putOrganization(ctx, tx, o, v); err != nil {
 		return &influxdb.Error{
 			Err: err,
 		}
 	}
-	return nil
+
+	uid, _ := icontext.GetUserID(ctx)
+	return s.audit.Log(resource.Change{
+		Type:           resource.Create,
+		ResourceID:     o.ID,
+		ResourceType:   influxdb.OrgsResourceType,
+		OrganizationID: o.ID,
+		UserID:         uid,
+		ResourceBody:   v,
+		Time:           time.Now(),
+	})
 }
 
 func (s *Service) generateOrgID(ctx context.Context, tx Tx) (influxdb.ID, error) {
@@ -307,22 +323,30 @@ func (s *Service) generateOrgID(ctx context.Context, tx Tx) (influxdb.ID, error)
 
 // PutOrganization will put a organization without setting an ID.
 func (s *Service) PutOrganization(ctx context.Context, o *influxdb.Organization) error {
-	var err error
 	return s.kv.Update(ctx, func(tx Tx) error {
-		if pe := s.putOrganization(ctx, tx, o); pe != nil {
-			err = pe
+		v, err := json.Marshal(o)
+		if err != nil {
+			return influxdb.ErrInternalOrgServiceError(influxdb.OpPutOrganization, err)
 		}
-		return err
+
+		if err := s.putOrganization(ctx, tx, o, v); err != nil {
+			return err
+		}
+
+		uid, _ := icontext.GetUserID(ctx)
+		return s.audit.Log(resource.Change{
+			Type:           resource.Put,
+			ResourceID:     o.ID,
+			ResourceType:   influxdb.OrgsResourceType,
+			OrganizationID: o.ID,
+			UserID:         uid,
+			ResourceBody:   v,
+			Time:           time.Now(),
+		})
 	})
 }
 
-func (s *Service) putOrganization(ctx context.Context, tx Tx, o *influxdb.Organization) error {
-	v, err := json.Marshal(o)
-	if err != nil {
-		return &influxdb.Error{
-			Err: err,
-		}
-	}
+func (s *Service) putOrganization(ctx context.Context, tx Tx, o *influxdb.Organization, v []byte) error {
 	encodedID, err := o.ID.Encode()
 	if err != nil {
 		return &influxdb.Error{
@@ -452,10 +476,29 @@ func (s *Service) updateOrganization(ctx context.Context, tx Tx, id influxdb.ID,
 		}
 	}
 
-	if pe := s.putOrganization(ctx, tx, o); pe != nil {
+	v, err := json.Marshal(o)
+	if err != nil {
+		return nil, influxdb.ErrInternalOrgServiceError(influxdb.OpUpdateOrganization, err)
+	}
+	if pe := s.putOrganization(ctx, tx, o, v); pe != nil {
 		return nil, pe
 	}
 
+	uid, _ := icontext.GetUserID(ctx)
+	if err := s.audit.Log(resource.Change{
+		Type:           resource.Update,
+		ResourceID:     o.ID,
+		ResourceType:   influxdb.OrgsResourceType,
+		OrganizationID: o.ID,
+		UserID:         uid,
+		ResourceBody:   v,
+		Time:           time.Now(),
+	}); err != nil {
+		return nil, &influxdb.Error{
+			Err: err,
+		}
+	}
+
 	return o, nil
 }
 
@@ -484,7 +527,16 @@ func (s *Service) DeleteOrganization(ctx context.Context, id influxdb.ID) error
 		if pe := s.deleteOrganization(ctx, tx, id); pe != nil {
 			return pe
 		}
-		return nil
+
+		uid, _ := icontext.GetUserID(ctx)
+		return s.audit.Log(resource.Change{
+			Type:           resource.Delete,
+			ResourceID:     id,
+			ResourceType:   influxdb.OrgsResourceType,
+			OrganizationID: id,
+			UserID:         uid,
+			Time:           time.Now(),
+		})
 	})
 	if err != nil {
 		return &influxdb.Error{