OpenSearch Multi-Tenancy Setup#

Multi-tenancy in OpenSearch enables organizations to serve multiple customers or teams from a single cluster while maintaining strict data isolation and security boundaries. This guide provides a complete implementation for production-ready multi-tenant OpenSearch deployments.

Multi-Tenancy Architecture Overview#

OpenSearch supports multiple approaches to multi-tenancy, each with different trade-offs in terms of isolation, performance, and resource utilization:

1
graph TB
2
    subgraph "Multi-Tenancy Models"
3
        subgraph "Index-per-Tenant"
4
            T1_Index[Tenant1 Index]
5
            T2_Index[Tenant2 Index]
6
            T3_Index[Tenant3 Index]
7
        end
8

9
        subgraph "Alias-per-Tenant"
10
            Shared_Index[Shared Index]
11
            T1_Alias[Tenant1 Alias]
12
            T2_Alias[Tenant2 Alias]
13
            T3_Alias[Tenant3 Alias]
14
        end
15

16
        subgraph "Cluster-per-Tenant"
17
            T1_Cluster[Tenant1 Cluster]
18
            T2_Cluster[Tenant2 Cluster]
19
            T3_Cluster[Tenant3 Cluster]
20
        end
21
    end
22

23
    subgraph "Security Layer"
24
        Auth[Authentication]
25
        RBAC[Role-Based Access]
26
        FLS[Field Level Security]
27
        DLS[Document Level Security]
28
    end
29

30
    subgraph "Resource Management"
31
        ISM[Index State Management]
32
        Quotas[Resource Quotas]
33
        Monitoring[Tenant Monitoring]
34
    end
35

36
    T1_Alias --> Shared_Index
37
    T2_Alias --> Shared_Index
38
    T3_Alias --> Shared_Index
39

40
    Auth --> RBAC
41
    RBAC --> FLS
42
    RBAC --> DLS
43

44
    style T1_Index fill:#f96,stroke:#333,stroke-width:2px
45
    style RBAC fill:#9f9,stroke:#333,stroke-width:2px

Index-per-Tenant Implementation#

The index-per-tenant model provides the strongest isolation between tenants:

Index Template Configuration#

1
PUT _index_template/tenant-template
2
{
3
  "index_patterns": ["tenant-*"],
4
  "priority": 100,
5
  "template": {
6
    "settings": {
7
      "number_of_shards": 3,
8
      "number_of_replicas": 1,
9
      "index.refresh_interval": "5s",
10
      "index.max_result_window": 10000,
11
      "index.codec": "best_compression",
12
      "index.routing.allocation.total_shards_per_node": 2
13
    },
14
    "mappings": {
15
      "dynamic": "strict",
16
      "properties": {
17
        "tenant_id": {
18
          "type": "keyword",
19
          "index": false
20
        },
21
        "created_at": {
22
          "type": "date",
23
          "format": "epoch_millis"
24
        },
25
        "updated_at": {
26
          "type": "date",
27
          "format": "epoch_millis"
28
        },
29
        "data": {
30
          "type": "object",
31
          "enabled": true
32
        }
33
      }
34
    },
35
    "aliases": {}
36
  },
37
  "composed_of": ["component-template-security", "component-template-lifecycle"]
38
}

Tenant Provisioning Script#

1
#!/usr/bin/env python3
2
import os
3
import json
4
import hashlib
5
import requests
6
from datetime import datetime
7
from opensearchpy import OpenSearch
8

9
class TenantProvisioner:
10
    def __init__(self, host='localhost', port=9200, auth=('admin', 'admin')):
11
        self.client = OpenSearch(
12
            hosts=[{'host': host, 'port': port}],
13
            http_auth=auth,
14
            use_ssl=True,
15
            verify_certs=False,
16
            ssl_show_warn=False
17
        )
18
        self.security_api_url = f"https://{host}:{port}/_plugins/_security/api"
19
        self.auth = auth
20

21
    def create_tenant(self, tenant_id, tenant_name, admin_user):
22
        """Create a new tenant with all necessary configurations"""
23
        print(f"Creating tenant: {tenant_id}")
24

25
        # Create tenant index
26
        index_name = f"tenant-{tenant_id}"
27
        self._create_index(index_name, tenant_id)
28

29
        # Create tenant roles
30
        self._create_tenant_roles(tenant_id)
31

32
        # Create tenant admin user
33
        self._create_tenant_admin(tenant_id, admin_user)
34

35
        # Create index lifecycle policy
36
        self._create_lifecycle_policy(tenant_id)
37

38
        # Create monitoring dashboards
39
        self._create_monitoring_resources(tenant_id)
40

41
        print(f"Tenant {tenant_id} created successfully")
42

43
    def _create_index(self, index_name, tenant_id):
44
        """Create tenant-specific index"""
45
        body = {
46
            "settings": {
47
                "index.blocks.read_only_allow_delete": False,
48
                "analysis": {
49
                    "analyzer": {
50
                        "tenant_analyzer": {
51
                            "type": "custom",
52
                            "tokenizer": "standard",
53
                            "filter": ["lowercase", "stop", "snowball"]
54
                        }
55
                    }
56
                }
57
            },
58
            "aliases": {
59
                f"{tenant_id}-current": {},
60
                f"{tenant_id}-search": {
61
                    "filter": {
62
                        "term": {"tenant_id": tenant_id}
63
                    }
64
                }
65
            }
66
        }
67

68
        self.client.indices.create(index=index_name, body=body)
69

70
    def _create_tenant_roles(self, tenant_id):
71
        """Create roles for tenant access control"""
72

73
        # Tenant admin role
74
        admin_role = {
75
            "cluster_permissions": [
76
                "cluster_monitor",
77
                "indices_monitor"
78
            ],
79
            "index_permissions": [{
80
                "index_patterns": [f"tenant-{tenant_id}*"],
81
                "allowed_actions": [
82
                    "crud",
83
                    "create_index",
84
                    "manage",
85
                    "manage_aliases",
86
                    "delete",
87
                    "index",
88
                    "read",
89
                    "write",
90
                    "search"
91
                ]
92
            }],
93
            "tenant_permissions": [{
94
                "tenant_patterns": [tenant_id],
95
                "allowed_actions": ["kibana_all_write"]
96
            }]
97
        }
98

99
        # Tenant read-only role
100
        readonly_role = {
101
            "cluster_permissions": ["cluster_monitor"],
102
            "index_permissions": [{
103
                "index_patterns": [f"tenant-{tenant_id}*"],
104
                "allowed_actions": ["read", "search"]
105
            }],
106
            "tenant_permissions": [{
107
                "tenant_patterns": [tenant_id],
108
                "allowed_actions": ["kibana_all_read"]
109
            }]
110
        }
111

112
        # Create roles via Security API
113
        self._security_api_put(f"roles/{tenant_id}_admin", admin_role)
114
        self._security_api_put(f"roles/{tenant_id}_readonly", readonly_role)
115

116
    def _create_tenant_admin(self, tenant_id, admin_user):
117
        """Create tenant administrator user"""
118
        user_data = {
119
            "password": self._generate_password(tenant_id),
120
            "opendistro_security_roles": [f"{tenant_id}_admin"],
121
            "backend_roles": [f"tenant_{tenant_id}"],
122
            "attributes": {
123
                "tenant_id": tenant_id,
124
                "created_at": datetime.now().isoformat()
125
            }
126
        }
127

128
        self._security_api_put(f"internalusers/{admin_user}", user_data)
129

130
    def _create_lifecycle_policy(self, tenant_id):
131
        """Create index lifecycle management policy"""
132
        policy = {
133
            "policy": {
134
                "description": f"Lifecycle policy for tenant {tenant_id}",
135
                "default_state": "hot",
136
                "states": [
137
                    {
138
                        "name": "hot",
139
                        "actions": [
140
                            {
141
                                "rollover": {
142
                                    "min_index_age": "7d",
143
                                    "min_size": "50gb"
144
                                }
145
                            }
146
                        ],
147
                        "transitions": [
148
                            {
149
                                "state_name": "warm",
150
                                "conditions": {
151
                                    "min_index_age": "30d"
152
                                }
153
                            }
154
                        ]
155
                    },
156
                    {
157
                        "name": "warm",
158
                        "actions": [
159
                            {
160
                                "replica_count": {
161
                                    "number_of_replicas": 1
162
                                }
163
                            },
164
                            {
165
                                "shrink": {
166
                                    "number_of_shards": 1
167
                                }
168
                            }
169
                        ],
170
                        "transitions": [
171
                            {
172
                                "state_name": "delete",
173
                                "conditions": {
174
                                    "min_index_age": "90d"
175
                                }
176
                            }
177
                        ]
178
                    },
179
                    {
180
                        "name": "delete",
181
                        "actions": [
182
                            {
183
                                "delete": {}
184
                            }
185
                        ]
186
                    }
187
                ]
188
            }
189
        }
190

191
        response = self.client.transport.perform_request(
192
            'PUT',
193
            f'/_plugins/_ism/policies/{tenant_id}_lifecycle',
194
            body=policy
195
        )
196

197
    def _create_monitoring_resources(self, tenant_id):
198
        """Create monitoring dashboards and visualizations"""
199
        # Create index pattern
200
        pattern_id = f"{tenant_id}-pattern"
201
        pattern_body = {
202
            "title": f"tenant-{tenant_id}*",
203
            "timeFieldName": "created_at"
204
        }
205

206
        # This would integrate with OpenSearch Dashboards API
207
        # Implementation depends on Dashboards version and configuration
208

209
    def _generate_password(self, tenant_id):
210
        """Generate secure password for tenant"""
211
        # In production, use a secure password generator
212
        return hashlib.sha256(f"{tenant_id}-{datetime.now()}".encode()).hexdigest()[:16]
213

214
    def _security_api_put(self, endpoint, data):
215
        """Make PUT request to Security API"""
216
        response = requests.put(
217
            f"{self.security_api_url}/{endpoint}",
218
            json=data,
219
            auth=self.auth,
220
            verify=False
221
        )
222
        response.raise_for_status()
223
        return response.json()
224

225
# Usage example
226
if __name__ == "__main__":
227
    provisioner = TenantProvisioner()
228
    provisioner.create_tenant("acme-corp", "ACME Corporation", "acme_admin")

Document-Level Security Implementation#

For shared index multi-tenancy, document-level security (DLS) provides data isolation:

1
graph LR
2
    subgraph "Shared Index"
3
        Doc1[Document 1<br/>tenant: A]
4
        Doc2[Document 2<br/>tenant: B]
5
        Doc3[Document 3<br/>tenant: A]
6
        Doc4[Document 4<br/>tenant: C]
7
    end
8

9
    subgraph "DLS Filters"
10
        FilterA[Tenant A Filter<br/>tenant_id: A]
11
        FilterB[Tenant B Filter<br/>tenant_id: B]
12
        FilterC[Tenant C Filter<br/>tenant_id: C]
13
    end
14

15
    subgraph "User Views"
16
        UserA[User A<br/>Sees: Doc1, Doc3]
17
        UserB[User B<br/>Sees: Doc2]
18
        UserC[User C<br/>Sees: Doc4]
19
    end
20

21
    FilterA --> Doc1
22
    FilterA --> Doc3
23
    FilterB --> Doc2
24
    FilterC --> Doc4
25

26
    Doc1 --> UserA
27
    Doc3 --> UserA
28
    Doc2 --> UserB
29
    Doc4 --> UserC
30

31
    style FilterA fill:#f96,stroke:#333,stroke-width:2px
32
    style UserA fill:#9f9,stroke:#333,stroke-width:2px

DLS Role Configuration#

1
PUT _plugins/_security/api/roles/tenant_dls_role
2
{
3
  "cluster_permissions": ["cluster_monitor"],
4
  "index_permissions": [{
5
    "index_patterns": ["shared-data-*"],
6
    "dls": "{\"term\": {\"tenant_id\": \"${attr.internal.tenant_id}\"}}",
7
    "allowed_actions": [
8
      "read",
9
      "write",
10
      "delete",
11
      "search",
12
      "index"
13
    ]
14
  }]
15
}

Field-Level Security Configuration#

1
PUT _plugins/_security/api/roles/tenant_fls_role
2
{
3
  "cluster_permissions": ["cluster_monitor"],
4
  "index_permissions": [{
5
    "index_patterns": ["shared-data-*"],
6
    "dls": "{\"term\": {\"tenant_id\": \"${attr.internal.tenant_id}\"}}",
7
    "fls": [
8
      "~internal_*",
9
      "~system_*",
10
      "~admin_notes"
11
    ],
12
    "allowed_actions": ["read", "search"]
13
  }]
14
}

Kibana/OpenSearch Dashboards Multi-Tenancy#

Tenant Space Configuration#

1
opensearch.username: "kibanaserver"
2
opensearch.password: "kibanaserver"
3
opensearch.requestHeadersWhitelist: ["securitytenant", "Authorization"]
4

5
opensearch_security.multitenancy.enabled: true
6
opensearch_security.multitenancy.tenants.enable_global: true
7
opensearch_security.multitenancy.tenants.enable_private: true
8
opensearch_security.multitenancy.tenants.preferred: ["Private", "Global"]
9
opensearch_security.readonly_mode.roles: ["readonly_role"]
10

11
# Tenant branding
12
opensearch_security.multitenancy.custom_branding:
13
  tenant_a:
14
    logo: "/assets/tenant_a_logo.svg"
15
    favicon: "/assets/tenant_a_favicon.ico"
16
    title: "Tenant A Analytics"
17
  tenant_b:
18
    logo: "/assets/tenant_b_logo.svg"
19
    favicon: "/assets/tenant_b_favicon.ico"
20
    title: "Tenant B Dashboard"

Automated Tenant Dashboard Setup#

1
#!/usr/bin/env python3
2
import requests
3
import json
4
from typing import Dict, List
5

6
class DashboardManager:
7
    def __init__(self, dashboards_url: str, auth: tuple):
8
        self.url = dashboards_url
9
        self.auth = auth
10
        self.headers = {
11
            'Content-Type': 'application/json',
12
            'osd-xsrf': 'true'
13
        }
14

15
    def create_tenant_space(self, tenant_id: str):
16
        """Create a new tenant space in OpenSearch Dashboards"""
17

18
        # Create saved objects for tenant
19
        self._create_index_pattern(tenant_id)
20
        self._create_default_dashboard(tenant_id)
21
        self._create_default_visualizations(tenant_id)
22

23
    def _create_index_pattern(self, tenant_id: str):
24
        """Create index pattern for tenant"""
25
        pattern = {
26
            "attributes": {
27
                "title": f"tenant-{tenant_id}*",
28
                "timeFieldName": "created_at",
29
                "fields": "[]",
30
                "fieldFormatMap": "{}"
31
            }
32
        }
33

34
        response = requests.post(
35
            f"{self.url}/api/saved_objects/index-pattern",
36
            headers={**self.headers, 'securitytenant': tenant_id},
37
            json=pattern,
38
            auth=self.auth,
39
            verify=False
40
        )
41

42
        return response.json()
43

44
    def _create_default_dashboard(self, tenant_id: str):
45
        """Create default dashboard for tenant"""
46
        dashboard = {
47
            "attributes": {
48
                "title": f"{tenant_id} Overview Dashboard",
49
                "hits": 0,
50
                "description": f"Main dashboard for {tenant_id}",
51
                "panelsJSON": json.dumps([
52
                    {
53
                        "id": "1",
54
                        "type": "visualization",
55
                        "gridData": {
56
                            "x": 0,
57
                            "y": 0,
58
                            "w": 24,
59
                            "h": 15
60
                        }
61
                    },
62
                    {
63
                        "id": "2",
64
                        "type": "visualization",
65
                        "gridData": {
66
                            "x": 24,
67
                            "y": 0,
68
                            "w": 24,
69
                            "h": 15
70
                        }
71
                    }
72
                ]),
73
                "optionsJSON": json.dumps({
74
                    "darkTheme": False,
75
                    "hidePanelTitles": False,
76
                    "useMargins": True
77
                }),
78
                "version": 1,
79
                "timeRestore": True,
80
                "timeTo": "now",
81
                "timeFrom": "now-7d",
82
                "refreshInterval": {
83
                    "pause": True,
84
                    "value": 0
85
                },
86
                "kibanaSavedObjectMeta": {
87
                    "searchSourceJSON": json.dumps({
88
                        "query": {
89
                            "language": "kuery",
90
                            "query": ""
91
                        },
92
                        "filter": []
93
                    })
94
                }
95
            }
96
        }
97

98
        response = requests.post(
99
            f"{self.url}/api/saved_objects/dashboard",
100
            headers={**self.headers, 'securitytenant': tenant_id},
101
            json=dashboard,
102
            auth=self.auth,
103
            verify=False
104
        )
105

106
        return response.json()
107

108
    def _create_default_visualizations(self, tenant_id: str):
109
        """Create default visualizations for tenant"""
110

111
        # Document count over time
112
        time_series_viz = {
113
            "attributes": {
114
                "title": f"{tenant_id} - Documents Over Time",
115
                "visState": json.dumps({
116
                    "title": f"{tenant_id} - Documents Over Time",
117
                    "type": "line",
118
                    "aggs": [
119
                        {
120
                            "id": "1",
121
                            "enabled": True,
122
                            "type": "count",
123
                            "params": {},
124
                            "schema": "metric"
125
                        },
126
                        {
127
                            "id": "2",
128
                            "enabled": True,
129
                            "type": "date_histogram",
130
                            "params": {
131
                                "field": "created_at",
132
                                "interval": "auto",
133
                                "min_doc_count": 0,
134
                                "extended_bounds": {}
135
                            },
136
                            "schema": "segment"
137
                        }
138
                    ]
139
                }),
140
                "uiStateJSON": "{}",
141
                "kibanaSavedObjectMeta": {
142
                    "searchSourceJSON": json.dumps({
143
                        "index": f"tenant-{tenant_id}*",
144
                        "query": {
145
                            "match_all": {}
146
                        }
147
                    })
148
                }
149
            }
150
        }
151

152
        response = requests.post(
153
            f"{self.url}/api/saved_objects/visualization",
154
            headers={**self.headers, 'securitytenant': tenant_id},
155
            json=time_series_viz,
156
            auth=self.auth,
157
            verify=False
158
        )
159

160
        return response.json()

Resource Isolation and Quotas#

Resource Allocation Strategy#

1
graph TB
2
    subgraph "Cluster Resources"
3
        CPU[CPU Cores: 32]
4
        Memory[Memory: 128GB]
5
        Storage[Storage: 2TB]
6
    end
7

8
    subgraph "Resource Pools"
9
        subgraph "Premium Tenants"
10
            PT_CPU[CPU: 16 cores]
11
            PT_Mem[Memory: 64GB]
12
            PT_Storage[Storage: 1TB]
13
        end
14

15
        subgraph "Standard Tenants"
16
            ST_CPU[CPU: 12 cores]
17
            ST_Mem[Memory: 48GB]
18
            ST_Storage[Storage: 750GB]
19
        end
20

21
        subgraph "Basic Tenants"
22
            BT_CPU[CPU: 4 cores]
23
            BT_Mem[Memory: 16GB]
24
            BT_Storage[Storage: 250GB]
25
        end
26
    end
27

28
    CPU --> PT_CPU
29
    CPU --> ST_CPU
30
    CPU --> BT_CPU
31

32
    Memory --> PT_Mem
33
    Memory --> ST_Mem
34
    Memory --> BT_Mem
35

36
    Storage --> PT_Storage
37
    Storage --> ST_Storage
38
    Storage --> BT_Storage
39

40
    style Premium fill:#f96,stroke:#333,stroke-width:2px
41
    style Standard fill:#99f,stroke:#333,stroke-width:2px
42
    style Basic fill:#9f9,stroke:#333,stroke-width:2px

Implementing Resource Quotas#

1
#!/usr/bin/env python3
2
import json
3
from opensearchpy import OpenSearch
4
from typing import Dict, Optional
5

6
class ResourceQuotaManager:
7
    def __init__(self, client: OpenSearch):
8
        self.client = client
9
        self.quota_index = ".tenant_quotas"
10

11
    def set_tenant_quota(self, tenant_id: str, quota_type: str = "standard"):
12
        """Set resource quota for a tenant"""
13

14
        quotas = {
15
            "premium": {
16
                "max_indices": 100,
17
                "max_shards": 500,
18
                "max_docs": 100000000,  # 100M documents
19
                "max_size_gb": 1000,
20
                "max_fields": 1000,
21
                "max_query_rate": 1000,  # queries per second
22
                "max_index_rate": 5000,  # docs per second
23
                "circuit_breaker": {
24
                    "request": "80%",
25
                    "total": "95%"
26
                }
27
            },
28
            "standard": {
29
                "max_indices": 50,
30
                "max_shards": 250,
31
                "max_docs": 50000000,  # 50M documents
32
                "max_size_gb": 500,
33
                "max_fields": 500,
34
                "max_query_rate": 500,
35
                "max_index_rate": 2500,
36
                "circuit_breaker": {
37
                    "request": "70%",
38
                    "total": "90%"
39
                }
40
            },
41
            "basic": {
42
                "max_indices": 10,
43
                "max_shards": 50,
44
                "max_docs": 10000000,  # 10M documents
45
                "max_size_gb": 100,
46
                "max_fields": 200,
47
                "max_query_rate": 100,
48
                "max_index_rate": 500,
49
                "circuit_breaker": {
50
                    "request": "60%",
51
                    "total": "85%"
52
                }
53
            }
54
        }
55

56
        quota = quotas.get(quota_type, quotas["basic"])
57
        quota["tenant_id"] = tenant_id
58
        quota["quota_type"] = quota_type
59
        quota["created_at"] = "now"
60

61
        # Store quota configuration
62
        self.client.index(
63
            index=self.quota_index,
64
            id=tenant_id,
65
            body=quota
66
        )
67

68
        # Apply index settings
69
        self._apply_index_settings(tenant_id, quota)
70

71
    def _apply_index_settings(self, tenant_id: str, quota: Dict):
72
        """Apply quota settings to tenant indices"""
73

74
        settings = {
75
            "index.max_result_window": min(quota["max_docs"], 10000),
76
            "index.max_regex_length": 1000,
77
            "index.max_terms_count": quota["max_fields"],
78
            "index.max_script_fields": 32,
79
            "index.requests.cache.enable": True,
80
            "index.queries.cache.enabled": True
81
        }
82

83
        # Apply to all tenant indices
84
        self.client.indices.put_settings(
85
            index=f"tenant-{tenant_id}*",
86
            body={"settings": settings}
87
        )
88

89
    def check_quota_usage(self, tenant_id: str) -> Dict:
90
        """Check current resource usage against quota"""
91

92
        # Get quota configuration
93
        quota_doc = self.client.get(index=self.quota_index, id=tenant_id)
94
        quota = quota_doc["_source"]
95

96
        # Get current usage
97
        stats = self.client.indices.stats(index=f"tenant-{tenant_id}*")
98

99
        total_docs = sum(idx["primaries"]["docs"]["count"]
100
                        for idx in stats["indices"].values())
101
        total_size = sum(idx["primaries"]["store"]["size_in_bytes"]
102
                        for idx in stats["indices"].values())
103
        total_shards = sum(len(idx["shards"])
104
                          for idx in stats["indices"].values())
105

106
        usage = {
107
            "tenant_id": tenant_id,
108
            "quota_type": quota["quota_type"],
109
            "usage": {
110
                "indices": {
111
                    "used": len(stats["indices"]),
112
                    "limit": quota["max_indices"],
113
                    "percentage": (len(stats["indices"]) / quota["max_indices"]) * 100
114
                },
115
                "shards": {
116
                    "used": total_shards,
117
                    "limit": quota["max_shards"],
118
                    "percentage": (total_shards / quota["max_shards"]) * 100
119
                },
120
                "documents": {
121
                    "used": total_docs,
122
                    "limit": quota["max_docs"],
123
                    "percentage": (total_docs / quota["max_docs"]) * 100
124
                },
125
                "storage_gb": {
126
                    "used": total_size / (1024**3),
127
                    "limit": quota["max_size_gb"],
128
                    "percentage": (total_size / (1024**3) / quota["max_size_gb"]) * 100
129
                }
130
            },
131
            "warnings": []
132
        }
133

134
        # Add warnings for high usage
135
        for metric, data in usage["usage"].items():
136
            if data["percentage"] > 80:
137
                usage["warnings"].append(f"{metric} usage above 80%")
138

139
        return usage

Query Rate Limiting#

Rate Limiter Implementation#

1
#!/usr/bin/env python3
2
import time
3
import redis
4
from functools import wraps
5
from typing import Optional, Callable
6

7
class TenantRateLimiter:
8
    def __init__(self, redis_host='localhost', redis_port=6379):
9
        self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
10

11
    def limit_queries(self, tenant_id: str, max_qps: int):
12
        """Decorator to enforce query rate limits per tenant"""
13
        def decorator(func: Callable):
14
            @wraps(func)
15
            def wrapper(*args, **kwargs):
16
                key = f"rate_limit:{tenant_id}:queries"
17

18
                try:
19
                    current = self.redis_client.incr(key)
20
                    if current == 1:
21
                        self.redis_client.expire(key, 1)
22

23
                    if current > max_qps:
24
                        raise RateLimitExceeded(
25
                            f"Tenant {tenant_id} exceeded query rate limit of {max_qps} QPS"
26
                        )
27

28
                    return func(*args, **kwargs)
29

30
                except redis.RedisError:
31
                    # If Redis is down, allow the query (fail open)
32
                    return func(*args, **kwargs)
33

34
            return wrapper
35
        return decorator
36

37
    def limit_indexing(self, tenant_id: str, max_docs_per_second: int):
38
        """Implement indexing rate limits using token bucket algorithm"""
39
        def decorator(func: Callable):
40
            @wraps(func)
41
            def wrapper(*args, **kwargs):
42
                bucket_key = f"token_bucket:{tenant_id}:indexing"
43
                timestamp_key = f"token_bucket:{tenant_id}:indexing:timestamp"
44

45
                # Token bucket parameters
46
                capacity = max_docs_per_second * 10  # Allow bursts
47
                refill_rate = max_docs_per_second
48

49
                pipe = self.redis_client.pipeline()
50

51
                # Get current tokens and last refill time
52
                pipe.get(bucket_key)
53
                pipe.get(timestamp_key)
54
                results = pipe.execute()
55

56
                current_tokens = int(results[0] or capacity)
57
                last_refill = float(results[1] or time.time())
58

59
                # Calculate tokens to add
60
                now = time.time()
61
                elapsed = now - last_refill
62
                tokens_to_add = int(elapsed * refill_rate)
63

64
                # Update bucket
65
                new_tokens = min(capacity, current_tokens + tokens_to_add)
66

67
                # Check if we have tokens available
68
                docs_count = kwargs.get('docs_count', 1)
69
                if new_tokens >= docs_count:
70
                    # Consume tokens
71
                    pipe = self.redis_client.pipeline()
72
                    pipe.set(bucket_key, new_tokens - docs_count)
73
                    pipe.set(timestamp_key, now)
74
                    pipe.expire(bucket_key, 60)
75
                    pipe.expire(timestamp_key, 60)
76
                    pipe.execute()
77

78
                    return func(*args, **kwargs)
79
                else:
80
                    raise RateLimitExceeded(
81
                        f"Tenant {tenant_id} exceeded indexing rate limit"
82
                    )
83

84
            return wrapper
85
        return decorator
86

87
class RateLimitExceeded(Exception):
88
    """Rate limit exceeded exception"""
89
    pass
90

91
# Usage example
92
rate_limiter = TenantRateLimiter()
93

94
@rate_limiter.limit_queries("tenant-123", max_qps=100)
95
def search_documents(query):
96
    # Perform search
97
    pass
98

99
@rate_limiter.limit_indexing("tenant-123", max_docs_per_second=1000)
100
def bulk_index(documents, docs_count=None):
101
    # Perform bulk indexing
102
    pass

Cross-Tenant Analytics#

Aggregated Metrics Collection#

1
#!/usr/bin/env python3
2
from opensearchpy import OpenSearch
3
from datetime import datetime, timedelta
4
import json
5

6
class CrossTenantAnalytics:
7
    def __init__(self, client: OpenSearch):
8
        self.client = client
9
        self.analytics_index = ".tenant_analytics"
10

11
    def collect_tenant_metrics(self):
12
        """Collect metrics across all tenants"""
13

14
        # Get all tenant indices
15
        all_indices = self.client.indices.get_alias(index="tenant-*")
16

17
        tenant_metrics = {}
18

19
        for index_name in all_indices:
20
            tenant_id = index_name.split("-")[1]
21

22
            if tenant_id not in tenant_metrics:
23
                tenant_metrics[tenant_id] = {
24
                    "tenant_id": tenant_id,
25
                    "timestamp": datetime.now().isoformat(),
26
                    "indices": [],
27
                    "total_docs": 0,
28
                    "total_size_bytes": 0,
29
                    "query_latency_ms": [],
30
                    "index_rate": 0,
31
                    "search_rate": 0
32
                }
33

34
            # Get index stats
35
            stats = self.client.indices.stats(index=index_name)
36
            idx_stats = stats["indices"][index_name]
37

38
            tenant_metrics[tenant_id]["indices"].append(index_name)
39
            tenant_metrics[tenant_id]["total_docs"] += idx_stats["primaries"]["docs"]["count"]
40
            tenant_metrics[tenant_id]["total_size_bytes"] += idx_stats["primaries"]["store"]["size_in_bytes"]
41

42
            # Get search stats
43
            search_stats = idx_stats["primaries"]["search"]
44
            tenant_metrics[tenant_id]["search_rate"] = search_stats.get("query_total", 0) / max(search_stats.get("query_time_in_millis", 1) / 1000, 1)
45

46
            # Get indexing stats
47
            indexing_stats = idx_stats["primaries"]["indexing"]
48
            tenant_metrics[tenant_id]["index_rate"] = indexing_stats.get("index_total", 0) / max(indexing_stats.get("index_time_in_millis", 1) / 1000, 1)
49

50
        # Store metrics
51
        for tenant_id, metrics in tenant_metrics.items():
52
            self.client.index(
53
                index=self.analytics_index,
54
                body=metrics
55
            )
56

57
        return tenant_metrics
58

59
    def generate_billing_report(self, start_date: datetime, end_date: datetime):
60
        """Generate billing report based on resource usage"""
61

62
        query = {
63
            "query": {
64
                "range": {
65
                    "timestamp": {
66
                        "gte": start_date.isoformat(),
67
                        "lte": end_date.isoformat()
68
                    }
69
                }
70
            },
71
            "aggs": {
72
                "tenants": {
73
                    "terms": {
74
                        "field": "tenant_id.keyword",
75
                        "size": 10000
76
                    },
77
                    "aggs": {
78
                        "avg_docs": {
79
                            "avg": {
80
                                "field": "total_docs"
81
                            }
82
                        },
83
                        "avg_storage_gb": {
84
                            "avg": {
85
                                "script": {
86
                                    "source": "doc['total_size_bytes'].value / (1024.0 * 1024.0 * 1024.0)"
87
                                }
88
                            }
89
                        },
90
                        "total_searches": {
91
                            "sum": {
92
                                "field": "search_rate"
93
                            }
94
                        },
95
                        "total_indexing": {
96
                            "sum": {
97
                                "field": "index_rate"
98
                            }
99
                        }
100
                    }
101
                }
102
            }
103
        }
104

105
        result = self.client.search(index=self.analytics_index, body=query)
106

107
        billing_report = []
108

109
        for bucket in result["aggregations"]["tenants"]["buckets"]:
110
            tenant_id = bucket["key"]
111

112
            # Calculate costs (example pricing model)
113
            storage_cost = bucket["avg_storage_gb"]["value"] * 0.10  # $0.10 per GB
114
            doc_cost = (bucket["avg_docs"]["value"] / 1000000) * 1.00  # $1.00 per million docs
115
            search_cost = (bucket["total_searches"]["value"] / 1000) * 0.01  # $0.01 per 1000 searches
116
            index_cost = (bucket["total_indexing"]["value"] / 10000) * 0.05  # $0.05 per 10k indexing ops
117

118
            total_cost = storage_cost + doc_cost + search_cost + index_cost
119

120
            billing_report.append({
121
                "tenant_id": tenant_id,
122
                "period": f"{start_date.date()} to {end_date.date()}",
123
                "usage": {
124
                    "avg_storage_gb": round(bucket["avg_storage_gb"]["value"], 2),
125
                    "avg_documents": int(bucket["avg_docs"]["value"]),
126
                    "total_searches": int(bucket["total_searches"]["value"]),
127
                    "total_indexing_ops": int(bucket["total_indexing"]["value"])
128
                },
129
                "costs": {
130
                    "storage": round(storage_cost, 2),
131
                    "documents": round(doc_cost, 2),
132
                    "searches": round(search_cost, 2),
133
                    "indexing": round(index_cost, 2),
134
                    "total": round(total_cost, 2)
135
                }
136
            })
137

138
        return billing_report

Security Best Practices#

Multi-Tenancy Security Checklist#

1
graph TB
2
    subgraph "Security Layers"
3
        subgraph "Network Security"
4
            TLS[TLS Encryption]
5
            Firewall[Firewall Rules]
6
            VPN[VPN Access]
7
        end
8

9
        subgraph "Authentication"
10
            SAML[SAML Integration]
11
            OIDC[OpenID Connect]
12
            MFA[Multi-Factor Auth]
13
        end
14

15
        subgraph "Authorization"
16
            RBAC[Role-Based Access]
17
            ABAC[Attribute-Based Access]
18
            DLS[Document-Level Security]
19
            FLS[Field-Level Security]
20
        end
21

22
        subgraph "Audit & Compliance"
23
            AuditLog[Audit Logging]
24
            Compliance[Compliance Monitoring]
25
            DataRetention[Data Retention]
26
        end
27
    end
28

29
    TLS --> SAML
30
    SAML --> RBAC
31
    RBAC --> AuditLog
32

33
    style TLS fill:#f96,stroke:#333,stroke-width:2px
34
    style RBAC fill:#9f9,stroke:#333,stroke-width:2px
35
    style AuditLog fill:#99f,stroke:#333,stroke-width:2px

Security Configuration Script#

1
#!/bin/bash
2
# Enable security features
3
cat > /etc/opensearch/opensearch.yml << EOF
4
# Security Configuration
5
plugins.security.ssl.transport.pemcert_filepath: node-cert.pem
6
plugins.security.ssl.transport.pemkey_filepath: node-key.pem
7
plugins.security.ssl.transport.pemtrustedcas_filepath: root-ca.pem
8
plugins.security.ssl.transport.enforce_hostname_verification: true
9
plugins.security.ssl.http.enabled: true
10
plugins.security.ssl.http.pemcert_filepath: node-cert.pem
11
plugins.security.ssl.http.pemkey_filepath: node-key.pem
12
plugins.security.ssl.http.pemtrustedcas_filepath: root-ca.pem
13

14
# Audit Configuration
15
plugins.security.audit.type: internal_opensearch
16
plugins.security.audit.config.log4j.logger_name: audit
17
plugins.security.audit.config.log4j.level: INFO
18
plugins.security.audit.config.disabled_rest_categories: NONE
19
plugins.security.audit.config.disabled_transport_categories: NONE
20

21
# Multi-tenancy
22
plugins.security.restapi.roles_enabled: ["all_access", "security_rest_api_access"]
23
plugins.security.check_snapshot_restore_write_privileges: true
24
plugins.security.enable_snapshot_restore_privilege: true
25

26
# DLS/FLS
27
plugins.security.dls_fls.enabled: true
28

29
# Field masking
30
plugins.security.compliance.history.write.log_diffs: true
31
plugins.security.compliance.history.read.watched_fields: ["personal_data.*", "sensitive.*"]
32
EOF
33

34
# Configure authentication backends
35
cat > /etc/opensearch/security/config.yml << EOF
36
_meta:
37
  type: "config"
38
  config_version: 2
39

40
config:
41
  dynamic:
42
    http:
43
      anonymous_auth_enabled: false
44
      xff:
45
        enabled: true
46
        internalProxies: '192\.168\.0\.0/16'
47

48
    authc:
49
      basic_internal_auth_domain:
50
        http_enabled: true
51
        transport_enabled: true
52
        order: 0
53
        http_authenticator:
54
          type: basic
55
          challenge: false
56
        authentication_backend:
57
          type: internal
58

59
      saml_auth_domain:
60
        http_enabled: true
61
        transport_enabled: false
62
        order: 1
63
        http_authenticator:
64
          type: saml
65
          challenge: true
66
          config:
67
            idp:
68
              metadata_file: saml-idp-metadata.xml
69
              entity_id: https://idp.company.com
70
            sp:
71
              entity_id: https://opensearch.company.com
72
              signature_algorithm: RSA_SHA256
73
            kibana_url: https://dashboards.company.com
74
        authentication_backend:
75
          type: noop
76

77
    authz:
78
      roles_from_myldap:
79
        http_enabled: true
80
        transport_enabled: true
81
        authorization_backend:
82
          type: ldap
83
          config:
84
            enable_ssl: true
85
            enable_start_tls: false
86
            enable_ssl_client_auth: false
87
            verify_hostnames: true
88
            hosts:
89
              - ldap.company.com:636
90
            bind_dn: cn=admin,dc=company,dc=com
91
            password: changeme
92
            userbase: ou=people,dc=company,dc=com
93
            usersearch: (uid={0})
94
            username_attribute: uid
95
            rolebase: ou=groups,dc=company,dc=com
96
            rolesearch: (member={0})
97
            userroleattribute: null
98
            userrolename: memberOf
99
            rolename: cn
100
            resolve_nested_roles: true
101
EOF
102

103
# Apply security configuration
104
/usr/share/opensearch/plugins/opensearch-security/tools/securityadmin.sh \
105
  -cd /etc/opensearch/security/ \
106
  -icl -nhnv \
107
  -cacert /etc/opensearch/root-ca.pem \
108
  -cert /etc/opensearch/admin-cert.pem \
109
  -key /etc/opensearch/admin-key.pem

Monitoring and Alerting#

Tenant-Specific Monitoring#

1
groups:
2
  - name: tenant_alerts
3
    interval: 30s
4
    rules:
5
      - alert: TenantQuotaExceeded
6
        expr: |
7
          (tenant_storage_used_bytes / tenant_storage_quota_bytes) > 0.9
8
        for: 5m
9
        labels:
10
          severity: warning
11
          team: platform
12
        annotations:
13
          summary: "Tenant {{ $labels.tenant_id }} approaching storage quota"
14
          description: "Tenant {{ $labels.tenant_id }} is using {{ $value | humanizePercentage }} of allocated storage"
15

16
      - alert: TenantHighQueryLatency
17
        expr: |
18
          histogram_quantile(0.95, tenant_query_latency_seconds) > 1
19
        for: 10m
20
        labels:
21
          severity: warning
22
          team: platform
23
        annotations:
24
          summary: "High query latency for tenant {{ $labels.tenant_id }}"
25
          description: "95th percentile query latency is {{ $value }}s"
26

27
      - alert: TenantIndexingRateAnomaly
28
        expr: |
29
          abs(rate(tenant_docs_indexed[5m]) - avg_over_time(rate(tenant_docs_indexed[5m])[1h:5m]))
30
          / avg_over_time(rate(tenant_docs_indexed[5m])[1h:5m]) > 2
31
        for: 15m
32
        labels:
33
          severity: info
34
          team: security
35
        annotations:
36
          summary: "Unusual indexing pattern for tenant {{ $labels.tenant_id }}"
37
          description: "Indexing rate deviates significantly from normal pattern"

Migration and Scaling#

Tenant Migration Strategy#

1
#!/usr/bin/env python3
2
import json
3
from opensearchpy import OpenSearch, helpers
4
from typing import List, Dict
5

6
class TenantMigrator:
7
    def __init__(self, source_client: OpenSearch, target_client: OpenSearch):
8
        self.source = source_client
9
        self.target = target_client
10

11
    def migrate_tenant(self, tenant_id: str, strategy: str = "reindex"):
12
        """Migrate tenant data between clusters or indices"""
13

14
        if strategy == "reindex":
15
            self._reindex_migration(tenant_id)
16
        elif strategy == "snapshot":
17
            self._snapshot_migration(tenant_id)
18
        elif strategy == "split":
19
            self._split_tenant(tenant_id)
20

21
    def _reindex_migration(self, tenant_id: str):
22
        """Migrate using reindex API"""
23

24
        source_index = f"tenant-{tenant_id}"
25
        target_index = f"tenant-{tenant_id}-new"
26

27
        # Create target index with updated settings
28
        self.target.indices.create(
29
            index=target_index,
30
            body={
31
                "settings": {
32
                    "number_of_shards": 5,
33
                    "number_of_replicas": 1,
34
                    "refresh_interval": "-1"  # Disable refresh during migration
35
                },
36
                "mappings": self.source.indices.get_mapping(index=source_index)[source_index]["mappings"]
37
            }
38
        )
39

40
        # Perform reindex
41
        self.target.reindex(
42
            body={
43
                "source": {
44
                    "remote": {
45
                        "host": "http://source-cluster:9200",
46
                        "username": "migration-user",
47
                        "password": "migration-password"
48
                    },
49
                    "index": source_index,
50
                    "size": 1000
51
                },
52
                "dest": {
53
                    "index": target_index
54
                }
55
            },
56
            wait_for_completion=False
57
        )
58

59
        # Monitor reindex progress
60
        task_id = response["task"]
61
        self._monitor_task(task_id)
62

63
        # Switch alias
64
        self.target.indices.update_aliases(
65
            body={
66
                "actions": [
67
                    {"remove": {"index": source_index, "alias": f"{tenant_id}-current"}},
68
                    {"add": {"index": target_index, "alias": f"{tenant_id}-current"}}
69
                ]
70
            }
71
        )
72

73
    def _split_tenant(self, tenant_id: str):
74
        """Split large tenant into multiple indices"""
75

76
        source_index = f"tenant-{tenant_id}"
77

78
        # Analyze data distribution
79
        distribution = self.source.search(
80
            index=source_index,
81
            body={
82
                "size": 0,
83
                "aggs": {
84
                    "data_categories": {
85
                        "terms": {
86
                            "field": "data_type.keyword",
87
                            "size": 100
88
                        }
89
                    }
90
                }
91
            }
92
        )
93

94
        # Create separate indices for each category
95
        for bucket in distribution["aggregations"]["data_categories"]["buckets"]:
96
            category = bucket["key"]
97
            target_index = f"tenant-{tenant_id}-{category}"
98

99
            # Create category-specific index
100
            self.target.indices.create(
101
                index=target_index,
102
                body={
103
                    "settings": {
104
                        "number_of_shards": 3,
105
                        "number_of_replicas": 1
106
                    }
107
                }
108
            )
109

110
            # Reindex category data
111
            self.target.reindex(
112
                body={
113
                    "source": {
114
                        "index": source_index,
115
                        "query": {
116
                            "term": {"data_type.keyword": category}
117
                        }
118
                    },
119
                    "dest": {
120
                        "index": target_index
121
                    }
122
                },
123
                wait_for_completion=False
124
            )

Conclusion#

Implementing multi-tenancy in OpenSearch requires careful planning and consideration of isolation requirements, performance characteristics, and operational complexity. The approach you choose—whether index-per-tenant, document-level security, or a hybrid model—should align with your specific use case, security requirements, and scalability needs.

Key takeaways for successful multi-tenant OpenSearch deployments:

Choose the right isolation model based on your security and performance requirements
Implement comprehensive access controls using OpenSearch Security features
Monitor and enforce resource quotas to prevent noisy neighbor problems
Automate tenant provisioning to ensure consistency and reduce operational overhead
Plan for growth with appropriate sharding strategies and migration paths
Implement thorough monitoring to track per-tenant metrics and ensure SLA compliance

With proper implementation, OpenSearch can efficiently serve thousands of tenants from a single cluster while maintaining strong isolation and performance guarantees.