1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
"""The Deployments cog helps with managing Kubernetes deployments."""
from http import HTTPStatus
from textwrap import dedent
from discord import ButtonStyle, Interaction, ui
from discord.ext import commands
from kubernetes_asyncio.client.models import V1Deployment
from kubernetes_asyncio.client.rest import ApiException
from tabulate import tabulate
from arthur.apis.kubernetes import deployments
from arthur.bot import KingArthur
from arthur.utils import generate_error_message
class ConfirmDeployment(ui.View):
"""A confirmation view for redeploying to Kubernetes."""
def __init__(self, author_id: int, deployment_ns: tuple[str, str]) -> None:
super().__init__()
self.confirmed = None
self.interaction = None
self.authorization = author_id
self.deployment = deployment_ns[1]
self.namespace = deployment_ns[0]
async def interaction_check(self, interaction: Interaction) -> bool:
"""Check the interactor is authorised."""
if interaction.user.id == self.authorization:
return True
await interaction.response.send_message(
generate_error_message(description="You are not authorized to perform this action."),
ephemeral=True,
)
return False
@ui.button(label="Confirm", style=ButtonStyle.green, row=0)
async def confirm(self, interaction: Interaction, _button: ui.Button) -> None:
"""Redeploy the specified service."""
try:
await deployments.restart_deployment(self.deployment, self.namespace)
except ApiException as e:
if e.status == HTTPStatus.NOT_FOUND:
return await interaction.message.edit(
content=generate_error_message(
description="Could not find deployment, check the namespace.",
),
view=None,
)
await interaction.message.edit(
content=generate_error_message(
description=f"Unexpected error occurred, error code {e.status}"
),
view=None,
)
else:
description = (
f":white_check_mark: Restarted deployment "
f"`{self.deployment}` in namespace `{self.namespace}`."
)
await interaction.message.edit(content=description, view=None)
self.stop()
return None
@ui.button(label="Cancel", style=ButtonStyle.grey, row=0)
async def cancel(self, interaction: Interaction, _button: ui.Button) -> None:
"""Logic for if the deployment is not approved."""
await interaction.message.edit(
content=":x: Redeployment aborted",
view=None,
)
self.stop()
def deployment_to_emote(deployment: V1Deployment) -> str:
"""Convert a deployment to an emote based on it's replica status."""
if deployment.status.available_replicas == deployment.spec.replicas:
return "\N{LARGE GREEN CIRCLE}"
if deployment.status.available_replicas == 0 or not deployment.status.available_replicas:
return "\N{LARGE RED CIRCLE}"
return "\N{LARGE YELLOW CIRCLE}"
class Deployments(commands.Cog):
"""Commands for working with Kubernetes Deployments."""
def __init__(self, bot: KingArthur) -> None:
self.bot = bot
@commands.group(name="deployments", aliases=["deploy"], invoke_without_command=True)
async def deployments(self, ctx: commands.Context) -> None:
"""Commands for working with Kubernetes Deployments."""
await ctx.send_help(ctx.command)
@deployments.command(name="list", aliases=["ls"])
async def deployments_list(self, ctx: commands.Context, namespace: str = "default") -> None:
"""List deployments in the selected namespace (defaults to default)."""
deploys = await deployments.list_deployments(namespace)
table_data = []
if len(deploys.items) == 0:
return await ctx.send(
generate_error_message(
description="No deployments found, check the namespace exists."
)
)
for deployment in deploys.items:
if deployment.status.available_replicas == deployment.spec.replicas:
emote = "\N{LARGE GREEN CIRCLE}"
elif (
deployment.status.available_replicas == 0
or not deployment.status.available_replicas
):
emote = "\N{LARGE RED CIRCLE}"
else:
emote = "\N{LARGE YELLOW CIRCLE}"
table_data.append(
[
emote,
deployment.metadata.name,
f"{deployment.status.available_replicas or 0}/{deployment.spec.replicas}",
]
)
table = tabulate(
table_data,
headers=["Status", "Deployment", "Replicas"],
tablefmt="psql",
colalign=("center", "left", "center"),
)
return_message = dedent("""
**Deployments in namespace `{0}`**
```
{1}
```
""")
await ctx.send(return_message.format(namespace, table))
return None
@deployments.command(name="restart", aliases=["redeploy"])
async def deployments_restart(
self, ctx: commands.Context, deployment: str, namespace: str = "default"
) -> None:
"""
Restart the specified deployment in the selected namespace (defaults to default).
It is also possible to use the more natural `kubectl` notation, that
is, specifying `rollout restart tooling/ff-bot` to restart the
deployment `ff-bot` in the namespace `tooling`.
"""
if "/" in deployment and namespace == "default":
deployment, namespace = deployment.split("/")
confirmation = ConfirmDeployment(ctx.author.id, [namespace, deployment])
msg = await ctx.send(
f":warning: Please confirm you want to restart `deploy/{deployment}` in `{namespace}`",
view=confirmation,
)
timed_out = await confirmation.wait()
if timed_out:
await msg.edit(
content=generate_error_message(
title="What is the airspeed velocity of Bella?",
description=(
"Whatever the answer may be, he is certainly "
"faster than you could select a confirmation option."
),
)
)
async def setup(bot: KingArthur) -> None:
"""Add the extension to the bot."""
await bot.add_cog(Deployments(bot))
|
