Skip to content

Instantly share code, notes, and snippets.

View lockwooddev's full-sized avatar
🐍
*productive hissing*

Carlo Smouter lockwooddev

🐍
*productive hissing*
View GitHub Profile
@lockwooddev
lockwooddev / update-index-default-pipeline.json
Created May 25, 2022 04:55
Update default_pipeline on legacy index
{
"index" : {
"default_pipeline": "metrics-forwarder"
}
}
@lockwooddev
lockwooddev / legacy-index-template.json
Created May 25, 2022 04:51
Legacy index template with default_pipeline set
{
"index_patterns": [
"legacy-index*"
],
"template": {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 1,
"default_pipeline": "metrics-forwarder",
}
{
"policy": {
"description": "Metrics retention policy",
"default_state": "hot",
"schema_version": 1,
"states": [
{
"name": "hot",
"actions": [
{
@lockwooddev
lockwooddev / datastream-index-template.json
Created May 23, 2022 14:48
data stream index template
{
"index_patterns": [
"metrics"
],
"data_stream": {},
"priority": 100,
"template": {
"settings": {
"number_of_shards": 2,
"number_of_replicas": 1
@lockwooddev
lockwooddev / ingest-pipeline.json
Created May 23, 2022 14:44
ingest pipeline for article
{
"description": "forward docs to data stream",
"processors": [
{
"set": {
"field": "_index",
"value": "metrics"
}
},
{
@pytest.mark.django_db
class TestWithVault:
def setup_method(self, method):
recursive_delete_secrets('startpath')
def teardown_method(self, method):
recursive_delete_secrets('startpath')
def test_yourtestcase(self, client):
type MockRobotsClient struct {
iter int64
Mocks []RobotRules
}
func NewMockRobotsClient(mocks []RobotRules) Roboteer {
return &MockRobotsClient{Mocks: mocks}
}
func (m *MockRobotsClient) FetchRobots(url string) (RobotRules, error) {
var robots = `
# Group 1
User-agent: Googlebot
Disallow: /nogooglebot/
# Group 2
User-agent: *
Allow: /
type Rule struct {
UserAgents []string
Disallow []string
Allow []string
}
func (r Rule) isEmpty() bool {
return (len(r.UserAgents) + len(r.Disallow) + len(r.Allow)) == 0
}
def recursive_delete_secrets(start_path):
paths = set()
def recursive_lookup(path):
try:
response = settings.VAULT_CLIENT.secrets.kv.v2.list_secrets(
path=path,
mount_point=settings.VAULT_MOUNTPOINT
)
except hvac.exceptions.InvalidPath: