מבוא

בזמן פריסת מערכת אחרת, עמדנו בפני הצורך לעבד מספר רב של יומנים שונים. ELK נבחר ככלי. מאמר זה ידון בניסיון שלנו בהגדרת מחסנית זו.

אנחנו לא מציבים יעד לתאר את כל היכולות שלו, אבל אנחנו רוצים להתרכז במיוחד בפתרון בעיות מעשיות. זה נובע מהעובדה שלמרות שיש כמות די גדולה של תיעוד ותמונות מוכנות, יש די הרבה מלכודות, לפחות מצאנו אותן.

פרסנו את המחסנית דרך docker-compose. יתרה מכך, היה לנו docker-compose.yml כתוב היטב, שאפשר לנו להעלות את הערימה כמעט ללא בעיות. ונראה לנו שהניצחון כבר קרוב, עכשיו נשנה אותו קצת כדי שיתאים לצרכים שלנו וזהו.

לרוע המזל, הניסיון להגדיר את המערכת לקבל ולעבד יומנים מהאפליקציה שלנו לא הצליח מיד. לכן החלטנו שכדאי ללמוד כל מרכיב בנפרד, ואז לחזור לקשריו.

אז, התחלנו עם logstash.

סביבה, פריסה, הפעלת Logstash בקונטיינר

לפריסה אנו משתמשים ב-docer-compose; הניסויים המתוארים כאן בוצעו על MacOS ו-Ubuntu 18.0.4.

תמונת ה-logstash שנרשמה ב-docker-compose.yml המקורי שלנו היא docker.elastic.co/logstash/logstash:6.3.2

נשתמש בו לניסויים.

כתבנו docker-compose.yml נפרד להפעלת logstash. כמובן שהיה אפשר להפעיל את התמונה משורת הפקודה, אבל פתרנו בעיה ספציפית, שבה אנחנו מריצים הכל מ-docker-compose.

בקצרה על קבצי תצורה

כדלקמן מהתיאור, ניתן להריץ את logstash עבור ערוץ אחד, ובמקרה זה הוא צריך להעביר את קובץ *.conf, או עבור מספר ערוצים, ובמקרה זה הוא צריך להעביר את הקובץ pipelines.yml, אשר בתורו , יקשר לקבצים .conf עבור כל ערוץ.
הלכנו בדרך השנייה. זה נראה לנו יותר אוניברסלי וניתן להרחבה. לכן יצרנו את pipelines.yml, ויצרנו ספריית pipelines בה נכניס קבצי .conf לכל ערוץ.

בתוך הקונטיינר יש קובץ תצורה נוסף - logstash.yml. אנחנו לא נוגעים בו, אנחנו משתמשים בו כפי שהוא.

אז, מבנה הספריות שלנו:

כדי לקבל נתוני קלט, לעת עתה אנו מניחים שזהו tcp ביציאה 5046, ולפלט נשתמש ב-stdout.

הנה תצורה פשוטה עבור ההשקה הראשונה. כי המשימה הראשונית היא להשיק.

אז יש לנו את docker-compose.yml הזה

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

מה אנחנו רואים כאן?

1. רשתות ונפחים נלקחו מהקובץ docker-compose.yml המקורי (זה שבו כל הערימה מושקת) ואני חושב שהם לא משפיעים מאוד על התמונה הכוללת כאן.
2. אנו יוצרים שירות(ים) אחד של logstash מהתמונה של docker.elastic.co/logstash/logstash:6.3.2 ונקרא לו logstash_one_channel.
3. אנו מעבירים את נמל 5046 בתוך המכולה, לאותו נמל פנימי.
4. אנו ממפים את קובץ תצורת הצינור שלנו ./config/pipelines.yml לקובץ /usr/share/logstash/config/pipelines.yml בתוך הקונטיינר, שם logstash יאסוף אותו ויהפוך אותו לקריאה בלבד, ליתר ביטחון.
5. אנו ממפים את ספריית ./config/pipelines, שבה יש לנו קבצים עם הגדרות ערוצים, לספריית /usr/share/logstash/config/pipelines וגם הופכים אותה לקריאה בלבד.

קובץ Pipelines.yml

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

ערוץ אחד עם מזהה HABR והנתיב לקובץ התצורה שלו מתוארים כאן.

ולבסוף הקובץ "./config/pipelines/habr_pipeline.conf"

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

בוא לא ניכנס לתיאור שלו לעת עתה, בואו ננסה להפעיל אותו:

docker-compose up

מה אנחנו רואים?

המכולה התחילה. אנחנו יכולים לבדוק את פעולתו:

echo '13123123123123123123123213123213' | nc localhost 5046

ואנחנו רואים את התגובה בקונסולת המכולה:

אבל במקביל, אנו רואים גם:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] לא ניתן לאחזר מידע רישיון משרת הרישיונות {:message=>“Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore ::ResolutionFailure] elasticsearch", ...

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] הצינור התחיל בהצלחה {:pipeline_id=>".monitoring-logstash", :thread=>"# "}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] צינורות הפועלים {:count=>2, :running_pipelines=>[:HABR, :.monitoring-logstash"], :non_running_pipelines=>[ ]}
logstash_one_channel | [2019-04-29T11:29:00,015][ERROR][logstash.inputs.metrics] X-Pack מותקן על Logstash אך לא על Elasticsearch. אנא התקן את X-Pack על Elasticsearch כדי להשתמש בתכונת הניטור. תכונות אחרות עשויות להיות זמינות.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] נקודת הקצה של Logstash API הופעלה בהצלחה {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] הפעלת בדיקת תקינות כדי לראות אם חיבור Elasticsearch פועל {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] ניסה להחיות את החיבור למופע ES מת, אך קיבל שגיאה. {:url=>“אלסטיקה:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch"}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] הפעלת בדיקת תקינות כדי לראות אם חיבור Elasticsearch פועל {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] ניסה להחיות את החיבור למופע ES מת, אך קיבל שגיאה. {:url=>“אלסטיקה:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch"}

והיומן שלנו זוחל כל הזמן.

כאן הדגשתי בירוק את ההודעה שהצינור הושק בהצלחה, באדום את הודעת השגיאה ובצהוב את ההודעה על ניסיון ליצור קשר אלסטיקה: 9200.
זה קורה מכיוון ש-logstash.conf, הכלול בתמונה, מכיל בדיקה לזמינות של elasticsearch. אחרי הכל, logstash מניח שהוא עובד כחלק מחסנית האיילים, אבל הפרדנו אותו.

אפשר לעבוד, אבל זה לא נוח.

הפתרון הוא להשבית בדיקה זו באמצעות משתנה הסביבה XPACK_MONITORING_ENABLED.

בואו נבצע שינוי ב-docker-compose.yml ונפעיל אותו שוב:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

עכשיו, הכל בסדר. המיכל מוכן לניסויים.

נוכל להקליד שוב בקונסולה הבאה:

echo '13123123123123123123123213123213' | nc localhost 5046

ולראות:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

עבודה בתוך ערוץ אחד

אז השקנו. עכשיו אתה באמת יכול לקחת את הזמן כדי להגדיר את logstash עצמו. בוא לא ניגע בקובץ pipelines.yml לעת עתה, בוא נראה מה נוכל להשיג על ידי עבודה עם ערוץ אחד.

אני חייב לומר שהעיקרון הכללי של עבודה עם קובץ תצורת הערוץ מתואר היטב במדריך הרשמי, כאן כאן
אם אתה רוצה לקרוא ברוסית, השתמשנו בזה מאמר(אבל תחביר השאילתה שם ישן, אנחנו צריכים לקחת זאת בחשבון).

בוא נעבור ברצף מקטע הקלט. כבר ראינו עבודה על TCP. מה עוד יכול להיות מעניין כאן?

בדוק הודעות באמצעות פעימות לב

יש הזדמנות כל כך מעניינת ליצור הודעות בדיקה אוטומטיות.
כדי לעשות זאת, עליך להפעיל את תוסף heartbean בקטע הקלט.

input {
  heartbeat {
    message => "HeartBeat!"
   }
  } 

הפעל אותו, התחל לקבל פעם בדקה

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

אם אנחנו רוצים לקבל לעתים קרובות יותר, אנחנו צריכים להוסיף את פרמטר המרווח.
כך נקבל הודעה כל 10 שניות.

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

אחזור נתונים מקובץ

החלטנו גם להסתכל על מצב הקובץ. אם זה עובד מצוין עם הקובץ, אז אולי אין צורך בסוכן, לפחות לשימוש מקומי.

לפי התיאור, מצב הפעולה צריך להיות דומה לזנב -f, כלומר. קורא שורות חדשות או, כאופציה, קורא את כל הקובץ.

אז מה אנחנו רוצים להשיג:

1. אנו רוצים לקבל שורות שמצורפות לקובץ יומן אחד.
2. אנו רוצים לקבל נתונים שנכתבים למספר קובצי יומן, תוך יכולת להפריד בין מה שמתקבל מאיפה.
3. אנו רוצים לוודא שכאשר יופעל מחדש את logstash, הוא לא יקבל את הנתונים האלה שוב.
4. אנחנו רוצים לבדוק שאם logstash כבוי, והנתונים ממשיכים להיכתב לקבצים, אז כשנריץ אותו, נקבל את הנתונים האלה.

כדי לבצע את הניסוי, בואו נוסיף שורה נוספת ל-docker-compose.yml, ונפתח את הספרייה שבה שמנו את הקבצים.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

ושנה את קטע הקלט ב-habr_pipeline.conf

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

בואו נתחיל:

docker-compose up

כדי ליצור ולכתוב קובצי יומן נשתמש בפקודה:


echo '1' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

כן, זה עובד!

יחד עם זאת, אנו רואים שהוספנו אוטומטית את שדה הנתיב. המשמעות היא שבעתיד נוכל לסנן רשומות לפיו.

בוא ננסה שוב:

echo '2' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

ועכשיו לקובץ אחר:

 echo '1' >> logs/number2.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

גדול! הקובץ נאסף, הנתיב צוין נכון, הכל בסדר.

עצור את ה-logstash והתחל מחדש. בוא נחכה. שתיקה. הָהֵן. אנחנו לא מקבלים את הרישומים האלה שוב.

ועכשיו הניסוי הנועז ביותר.

התקן את logstash והפעל:

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

הפעל שוב את logstash וראה:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

הידד! הכל נאסף.

אבל עלינו להזהיר אותך לגבי הדברים הבאים. אם המיכל עם logstash נמחק (docker stop logstash_one_channel && docker rm logstash_one_channel), אז שום דבר לא ייקלט. מיקום הקובץ שאליו הוא נקרא נשמר בתוך המיכל. אם תפעיל אותו מאפס, הוא יקבל רק שורות חדשות.

קריאת קבצים קיימים

נניח שאנחנו משיקים את logstash בפעם הראשונה, אבל כבר יש לנו יומנים ואנו רוצים לעבד אותם.
אם נריץ logstash עם קטע הקלט בו השתמשנו למעלה, לא נקבל כלום. רק שורות חדשות יעובדו על ידי logstash.

על מנת שניתן יהיה למשוך שורות מקבצים קיימים למעלה, עליך להוסיף שורה נוספת למקטע הקלט:

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

יתר על כן, יש ניואנס: זה משפיע רק על קבצים חדשים ש-logstash עדיין לא ראה. עבור אותם קבצים שכבר היו בשדה הראייה של logstash, הוא כבר זכר את גודלם וכעת ייקח בהם רק ערכים חדשים.

בואו נעצור כאן ונלמד את קטע הקלט. יש עדיין אפשרויות רבות, אבל זה מספיק לנו לניסויים נוספים לעת עתה.

ניתוב ושינוי נתונים

בוא ננסה לפתור את הבעיה הבאה, נניח שיש לנו הודעות מערוץ אחד, חלקן אינפורמטיביות וחלקן הודעות שגיאה. הם שונים לפי תג. חלקם הם INFO, אחרים הם שגיאה.

אנחנו צריכים להפריד אותם ביציאה. הָהֵן. אנו כותבים הודעות מידע בערוץ אחד, והודעות שגיאה בערוץ אחר.

כדי לעשות זאת, עבור מקטע הקלט לסינון ופלט.

באמצעות קטע הסינון, ננתח את ההודעה הנכנסת, ונקבל ממנה hash (צמדי מפתח-ערך), שאיתו כבר נוכל לעבוד, כלומר. לפרק לפי התנאים. ובמדור הפלט נבחר הודעות ונשלח כל אחת לערוץ שלה.

ניתוח הודעה עם גרוק

על מנת לנתח מחרוזות טקסט ולקבל מהן סט שדות, ישנו תוסף מיוחד בקטע הסינון - grok.

מבלי להציב לעצמי למטרה לתת תיאור מפורט שלו כאן (בשביל זה אני מתייחס תיעוד רשמי), אני אתן את הדוגמה הפשוטה שלי.

כדי לעשות זאת, עליך להחליט על הפורמט של מחרוזות הקלט. יש לי אותם ככה:

הודעת מידע אחת 1
2 הודעת שגיאה2

הָהֵן. המזהה בא קודם, ואז INFO/ERROR, ואז איזו מילה ללא רווחים.
זה לא קשה, אבל זה מספיק כדי להבין את עקרון הפעולה.

אז, בקטע הסינון של תוסף grok, עלינו להגדיר דפוס לניתוח המחרוזות שלנו.

זה ייראה כך:

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  } 

בעיקרו של דבר זה ביטוי רגולרי. נעשה שימוש בדפוסים מוכנים, כגון INT, LOGLEVEL, WORD. את התיאור שלהם, כמו גם דפוסים אחרים, ניתן למצוא כאן כאן

כעת, במעבר דרך המסנן הזה, המחרוזת שלנו תהפוך ל-hash של שלושה שדות: message_id, message_type, message_text.

הם יוצגו בקטע הפלט.

ניתוב הודעות למקטע הפלט באמצעות הפקודה if

בקטע הפלט, כזכור, עמדנו לפצל את ההודעות לשני זרמים. חלקם - שהם iNFO, ייצאו לקונסולה, ועם שגיאות, נוציא לקובץ.

איך נפריד בין ההודעות הללו? מצב הבעיה כבר מציע פתרון - אחרי הכל, כבר יש לנו שדה message_type ייעודי, שיכול לקחת רק שני ערכים: INFO ו-ERROR. על בסיס זה נעשה בחירה באמצעות הצהרת if.

if [message_type] == "ERROR" {
        # Здесь выводим в файл
       } else
     {
      # Здесь выводим в stdout
    }

ניתן למצוא תיאור של עבודה עם שדות ואופרטורים בחלק זה מדריך רשמי.

עכשיו, לגבי המסקנה עצמה.

פלט מסוף, הכל ברור כאן - stdout {}

אבל הפלט לקובץ - זכרו שאנחנו מריצים את כל זה ממיכל וכדי שהקובץ בו נכתוב את התוצאה יהיה נגיש מבחוץ, אנחנו צריכים לפתוח את הספרייה הזו ב-docker-compose.yml.

סך הכל:

קטע הפלט של הקובץ שלנו נראה כך:


output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "custom format: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

ב-docker-compose.yml אנו מוסיפים אמצעי אחסון נוסף לפלט:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

אנחנו משיקים אותו, מנסים אותו ורואים חלוקה לשני זרמים.

מקור: www.habr.com