Skip to content

Instantly share code, notes, and snippets.

import boto3
import uuid
import argparse
def assumed_role_session(role_arn):
sts_client = boto3.client('sts')
assumed_role_object=sts_client.assume_role(
RoleArn=role_arn,
RoleSessionName=uuid.uuid1().hex
import org.apache.commons.io.IOUtils
import uk.ac.wellcome.storage.streaming.Codec
import scala.util.{Success, Try}
trait IdempotentVersionedWritable[Id, V, T]
extends Writable[Version[Id, V], T]
with Readable[Version[Id, V], T] {
implicit val codec: Codec[T]
import uk.ac.wellcome.storage.RetryableError
import scala.util.Random
import scala.annotation.tailrec
object RetryOps {
implicit class Retry[In, Out, Error](f: In => Either[Error, Out]) {
def retry(maxAttempts: Int = 1): In => Either[Error, Out] =
(in: In) => retryInternal(maxAttempts)(in)
@kenoir
kenoir / show_queues.py
Created September 9, 2019 09:15
Graphically show SQS queue lengths
#!/usr/local/bin/python3
import datetime
import uuid
import argparse
import asciiplotlib as apl
import botocore
import boto3

Load Testing guidelines

Here are some guidelines established while trying to load test our platform.

Define your requirements up front

It's a good idea to start with a clear idea of where you want to get to. Define how many requests per second consitute a passing load test and lay those assertions down in code.

Don't load test your load tester

import boto3
import uuid
def create_session(role_arn):
client = boto3.client('sts')
response = client.assume_role(
RoleArn=role_arn,
RoleSessionName=str(uuid.uuid1())
)
@kenoir
kenoir / update_task_definition.py
Created August 29, 2019 16:51
Update an ECS task definition image only
def _get_attributes(target_dict, attributes):
return {attr: target_dict.get(attr) for attr in attributes if target_dict.get(attr)}
def _update_dict(target_dict, field, value):
target_dict[field] = value
return target_dict
def update_container_definitions(container_definitions, name, new_image):
update_matching = [_update_dict(definition, 'image', new_image)
for definition in container_definitions if definition['name'] == name]
#!/usr/bin/env bash
docker run \
--rm -it \
-v "$HOME/.aws:/root/.aws" \
-v "$(pwd):/project" mesosphere/aws-cli $@
#!/usr/bin/env bash
# ~/.azuresso
#
# {
# "AZURE_USER_ID": "user@domain"
# "AZURE_TENANT_ID" : "11111111-1111-1111-11111111111",
# "AZURE_APP_ID_URI" : "11111111-1111-1111-11111111111"
# }
@kenoir
kenoir / get_attributes.py
Last active August 28, 2019 08:40
Clean and filter a target dict
# Usage:
#
# target_dict = {
# 'foo': 'value'
# 'baz': 'another_value'
# }
#
# attributes = ['foo', 'bar']
# new_dict = get_attributes(target_dict
# assert new_dict == { 'foo': 'value' }